Tutorial: Procesamiento de eventos de Apache Kafka para Event Hubs mediante Stream Analytics

Artículo
06/01/2023

En este artículo se muestra cómo transmitir en secuencias datos a Event Hubs y cómo procesarlos con Azure Stream Analytics. Estos pasos son los siguientes:

Cree un espacio de nombres de Event Hubs.
Cree un cliente de Kafka que envíe mensajes al centro de eventos.
Cree un trabajo de Stream Analytics que copie datos del centro de eventos en Azure Blob Storage.

No es necesario cambiar a los clientes de protocolo ni ejecutar sus propios clústeres al usar el punto de conexión de Kafka expuesto por un centro de eventos. Azure Event Hubs admite Apache Kafka, versión 1.0. y superiores.

Prerrequisitos

Para completar esta guía de inicio rápido, asegúrese de cumplir los siguientes requisitos previos:

Suscripción a Azure. Si no tiene una, cree una cuenta gratuita antes de empezar.
Kit de desarrollo de Java (JDK) 1.7+.
Descargue e instale un archivo binario de Maven.
Git
Una cuenta de Azure Storage. Si no tiene, cree una antes de continuar. El trabajo de Stream Analytics en este tutorial almacena los datos de salida en Azure Blob Storage.

Creación de un espacio de nombres de Event Hubs

Al crear un espacio de nombres de Event Hubs, se habilita automáticamente el punto de conexión de Kafka para el espacio de nombres. Puede transmitir eventos a Event Hubs desde las aplicaciones que usan el protocolo de Kafka. Siga las instrucciones paso a paso de Creación de un centro de eventos mediante Azure Portal para crear un espacio de nombres de Event Hubs. Si usa un clúster dedicado, consulte Creación de un espacio de nombres y un centro de eventos en un clúster dedicado.

Nota:

Event Hubs para Kafka no se admite en el nivel básico.

Envío de mensajes con Kafka en Event Hubs

Clone el repositorio de Azure Event Hubs para Kafka en la máquina.
Vaya a la carpeta azure-event-hubs-for-kafka/quickstart/java/producer.

Actualice los detalles de configuración para el productor en src/main/resources/producer.config. Especifique el nombre y cadena de conexión para el espacio de nombres del centro de eventos.

bootstrap.servers={EVENT HUB NAMESPACE}.servicebus.windows.net:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{CONNECTION STRING for EVENT HUB NAMESPACE}";

Vaya a azure-event-hubs-for-kafka/quickstart/java/producer/src/main/java/ y abra el archivo TestDataReporter.java en el editor que prefiera.

Comente las siguientes líneas de código:

            //final ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(TOPIC, time, "Test Data " + i);

Agregue la siguiente línea de código en lugar del código con comentarios:
```
            final ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(TOPIC, time, "{ \"eventData\": \"Test Data " + i + "\" }");            
```
Este código envía los datos del evento en formato JSON. Al configurar la entrada para un trabajo de Stream Analytics, especifique JSON como formato de los datos de entrada.
Ejecute el productor y transmítalo a Event Hubs. En un equipo Windows, al usar un símbolo del sistema Node.js, cambie a la carpeta azure-event-hubs-for-kafka/quickstart/java/producer antes de ejecutar estos comandos.
```
mvn clean package
mvn exec:java -Dexec.mainClass="TestProducer"                                    
```

Compruebe que ese centro de eventos recibe los datos.

Seleccione Event Hubs en ENTIDADES. Confirme que ve un centro de eventos denominado prueba.
Confirme que ve los mensajes que entran en el centro de eventos.

Procesamiento de los datos de eventos mediante un trabajo de Stream Analytics

En esta sección, vamos a crear un trabajo de Azure Stream Analytics. El cliente de Kafka envía eventos al centro de eventos. Cree un trabajo de Stream Analytics que tome datos de eventos como entrada y los envíe a Azure Blob Storage. Si no tiene una cuenta de Azure Storage, cree una.

La consulta en el trabajo de Stream Analytics se pasa a través de los datos sin realizar ningún análisis. Puede crear una consulta que transforme los datos de entrada para generar datos de salida en un formato diferente o con información obtenida.

Creación de un trabajo de Stream Analytics

Seleccione + Crear un recurso en Azure Portal.
Seleccione Analytics en el menú Azure Marketplace y, luego, seleccione Stream Analytics job (Trabajo de Stream Analytics).
En la página Nuevo Stream Analytics, haga lo siguiente:
1. Escriba un nombre para el trabajo.
2. Seleccione su suscripción.
3. Seleccione Crear nuevo para el grupo de recursos y escriba el nombre. También puede usar un grupo de recursos existente.
4. Seleccione una ubicación para el trabajo.
5. Seleccione Crear para crear el trabajo.

Configuración de la entrada del trabajo

En el mensaje de notificación, seleccione Ir al recurso para ver la página Trabajo de Stream Analytics.
Seleccione Entradas en la sección TOPOLOGÍA DE TRABAJO en el menú izquierdo.
Seleccione Add stream input (Agregar entrada de flujo) y, luego, seleccione Event Hub (Centro de eventos).
En la página de configuración de la entrada del centro de eventos, haga lo siguiente:
1. Especifique un alias para la entrada.
2. Seleccione la suscripción de Azure.
3. Seleccione el espacio de nombres del centro de eventos que ha creado anteriormente.
4. Seleccione probar para el centro de eventos.
5. Seleccione Guardar.

Configuración de la salida del trabajo

Seleccione Salidas en la sección TOPOLOGÍA DE TRABAJO en el menú.
Seleccione + Agregar en la barra de herramientas y seleccione Blob Storage
Haga lo siguiente en la página de configuración de salida de Blob Storage:
1. Especifique un alias para la salida.
2. Selección la suscripciónde Azure.
3. Seleccione su cuenta de Azure Storage.
4. Escriba un nombre para el contenedor que almacena los datos de salida de la consulta de Stream Analytics.
5. Seleccione Guardar.

Definición de una consulta

Ya tiene un trabajo de Stream Analytics configurado para leer un flujo de datos de entrada. El siguiente paso consiste en crear una transformación que analice los datos en tiempo real. Definirá la consulta de transformación mediante el lenguaje de consulta de Stream Analytics. En este tutorial, definirá una consulta que se pasa a través de los datos sin realizar ninguna transformación.

Seleccione Consulta.
En la ventana de consulta, reemplace [YourOutputAlias] por el alias de salida que creó anteriormente.
Reemplace [YourInputAlias] por el alias de entrada que creó anteriormente.
Seleccione Guardar en la barra de herramientas.

Ejecución del trabajo de Stream Analytics

Seleccione Información general en el menú de la izquierda.
Seleccione Inicio.
En la página Iniciar trabajo, seleccione Iniciar.
Espere hasta que el estado del trabajo cambie de Iniciando a En ejecución.

Probar el escenario

Vuelva a ejecutar el productor de Kafka para enviar eventos al centro de eventos.

mvn exec:java -Dexec.mainClass="TestProducer"

Confirme que ve que los datos de salida se generan en Azure Blob Storage. Consulte un archivo JSON en el contenedor con 100 filas con un aspecto similar a las siguientes filas de ejemplo:

{"eventData":"Test Data 0","EventProcessedUtcTime":"2018-08-30T03:27:23.1592910Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}
{"eventData":"Test Data 1","EventProcessedUtcTime":"2018-08-30T03:27:23.3936511Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}
{"eventData":"Test Data 2","EventProcessedUtcTime":"2018-08-30T03:27:23.3936511Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}

El trabajo de Azure Stream Analytics recibió los datos de entrada del centro de eventos y los almacenó en Azure Blob Storage en este escenario.

Pasos siguientes

En este artículo se muestra cómo hacer streaming a instancias de Event Hubs sin cambiar los clientes del protocolo o ejecutar sus propios clústeres. Para más información sobre Event Hubs para Apache Kafka, consulte Guía del desarrollador de Apache Kafka para Azure Event Hubs.