Uso de Apache Flink con Azure Event Hubs para Apache Kafka

  • Artículo

En este tutorial se muestra cómo conectar Apache Flink a un centro de eventos sin cambiar los clientes del protocolo o ejecutar sus propios clústeres. Para más información sobre la compatibilidad de Event Hubs con el protocolo de consumidor de Apache Kafka, consulte Event Hubs para Apache Kafka.

En este tutorial, aprenderá a:

  • Creación de un espacio de nombres de Event Hubs
  • Clonación del proyecto de ejemplo
  • Ejecutar el productor Flink
  • Ejecutar el consumidor Flink

Nota

Este ejemplo está disponible en GitHub.

Prerrequisitos

Para completar este tutorial, asegúrese de cumplir estos requisitos previos:

  • Lea el artículo Event Hubs para Apache Kafka.
  • Suscripción a Azure. Si no tiene una, cree una cuenta gratuita antes de empezar.
  • Kit de desarrollo de Java (JDK) 1.7+
    • En Ubuntu, ejecute apt-get install default-jdk para instalar el JDK.
    • Asegúrese de establecer la variable de entorno JAVA_HOME para que apunte a la carpeta donde está instalado el JDK.
  • Descargue e instale un archivo binario de Maven
    • En Ubuntu, puede ejecutar apt-get install maven para instalar Maven.
  • Git
    • En Ubuntu, puede ejecutar sudo apt-get install git para instalar Git.

Creación de un espacio de nombres de Event Hubs

Se requiere un espacio de nombres de Event Hubs para enviar o recibir de cualquier servicio de Event Hubs. Vea Creación de un centro de eventos para obtener instrucciones sobre cómo crear un espacio de nombres y un centro de eventos. Asegúrese de copiar la cadena de conexión de Event Hubs para su uso posterior.

Clonación del proyecto de ejemplo

Ahora que tiene una cadena de conexión de Event Hubs, clone el repositorio de Azure Event Hubs para Kafka y vaya a la subcarpeta flink:

git clone https://github.com/Azure/azure-event-hubs-for-kafka.git
cd azure-event-hubs-for-kafka/tutorials/flink

Con el ejemplo de productor de Flink proporcionado, envíe mensajes al servicio Event Hubs.

Proporcionar un punto de conexión de Kafka para Event Hubs

Producer.config

Actualice los valores bootstrap.servers y sasl.jaas.config en producer/src/main/resources/producer.config para dirigir el productor al punto de conexión de Kafka para Event Hubs con la autenticación correcta.

bootstrap.servers={YOUR.EVENTHUBS.FQDN}:9093
client.id=FlinkExampleProducer
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
   username="$ConnectionString" \
   password="{YOUR.EVENTHUBS.CONNECTION.STRING}";

Importante

Reemplace {YOUR.EVENTHUBS.CONNECTION.STRING} por la cadena de conexión para el espacio de nombres de Event Hubs. Para obtener instrucciones sobre cómo obtener la cadena de conexión, consulte Obtención de una cadena de conexión de Event Hubs. A continuación se muestra un ejemplo de configuración: sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

Ejecución del productor desde la línea de comandos

Para ejecutar el productor desde la línea de comandos, genere los JAR y luego ejecútelo desde Maven (o genere los JAR con Maven, luego ejecútelo en Java al agregar los JAR de Kafka necesarios al parámetro classpath):

mvn clean package
mvn exec:java -Dexec.mainClass="FlinkTestProducer"

El productor ahora comenzará a enviar eventos al centro de eventos en el tema test e imprimir los eventos en stdout.

Con el ejemplo de consumidor proporcionado, reciba mensajes desde el centro de eventos.

Proporcionar un punto de conexión de Kafka para Event Hubs

consumer.config

Actualice los valores bootstrap.servers y sasl.jaas.config en consumer/src/main/resources/consumer.config para dirigir el consumidor al punto de conexión de Kafka para Event Hubs con la autenticación correcta.

bootstrap.servers={YOUR.EVENTHUBS.FQDN}:9093
group.id=FlinkExampleConsumer
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
   username="$ConnectionString" \
   password="{YOUR.EVENTHUBS.CONNECTION.STRING}";

Importante

Reemplace {YOUR.EVENTHUBS.CONNECTION.STRING} por la cadena de conexión para el espacio de nombres de Event Hubs. Para obtener instrucciones sobre cómo obtener la cadena de conexión, consulte Obtención de una cadena de conexión de Event Hubs. A continuación se muestra un ejemplo de configuración: sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

Ejecución del consumidor desde la línea de comandos

Para ejecutar el consumidor desde la línea de comandos, genere los JAR y luego ejecútelo desde Maven (o genere los JAR con Maven, luego ejecútelo en Java al agregar los JAR de Kafka necesarios al parámetro classpath):

mvn clean package
mvn exec:java -Dexec.mainClass="FlinkTestConsumer"

Si el centro de eventos tiene eventos (por ejemplo, si el productor también se está ejecutando), el consumidor comienza a recibir eventos del tema test.

Revise la guía del conector Kafka de Flink para más información acerca de cómo conectar Flink a Kafka.

Pasos siguientes

Para obtener más información acerca de Event Hubs para Kafka, consulte los artículos siguientes: