Ingesta de datos de Apache Kafka en Azure Data Explorer

Artículo
02/07/2024

Apache Kafka es una plataforma de streaming distribuida para crear canalizaciones de datos de streaming en tiempo real que mueven datos de forma confiable entre sistemas o aplicaciones. Kafka Connect es una herramienta para realizar streaming de datos de forma escalable y confiable entre Apache Kafka y otros sistemas de datos. El receptor de Kafka de Kusto actúa como conector de Kafka y no requiere el uso de código. Descargue el archivo jar del conector receptor desde el repositorio de Git o confluent Connector Hub.

En este artículo se muestra cómo ingerir datos con Kafka mediante una configuración independiente de Docker para simplificar la configuración del clúster de Kafka y del conector de Kafka.

Para más información, consulte el repositorio Git del conector y las especificaciones de la versión.

Requisitos previos

Suscripción a Azure. Cree una cuenta de Azure gratuita.
Un clúster y una base de datos de Azure Data Explorer con las directivas de retención y caché predeterminadas o una base de datos KQL en Microsoft Fabric.
Azure CLI.
Docker y Docker Compose.

Creación de una entidad de servicio de Microsoft Entra

La entidad de servicio Microsoft Entra se puede crear mediante el Azure Portal o mediante programación, como en el ejemplo siguiente.

Esta entidad de servicio será la identidad que usa el conector para escribir datos de la tabla en Kusto. Más adelante concederá permisos para que esta entidad de servicio acceda a los recursos de Kusto.

Inicie sesión en su suscripción de Azure a través de la CLI de Azure. A continuación, realice la autenticación en el explorador.
```
az login
```
Elija la suscripción para hospedar la entidad de seguridad. Este paso es necesario si tiene varias suscripciones.
```
az account set --subscription YOUR_SUBSCRIPTION_GUID
```

Cree la entidad de servicio. En este ejemplo, la entidad de servicio se llama my-service-principal.

az ad sp create-for-rbac -n "my-service-principal" --role Contributor --scopes /subscriptions/{SubID}

En los datos JSON devueltos, copie , appIdpasswordy tenant para su uso futuro.

{
  "appId": "1234abcd-e5f6-g7h8-i9j0-1234kl5678mn",
  "displayName": "my-service-principal",
  "name": "my-service-principal",
  "password": "1234abcd-e5f6-g7h8-i9j0-1234kl5678mn",
  "tenant": "1234abcd-e5f6-g7h8-i9j0-1234kl5678mn"
}

Ha creado una aplicación de Microsoft Entra y una entidad de servicio.

Creación de una tabla de destino

En el entorno de consulta, cree una tabla denominada Storms con el siguiente comando:

.create table Storms (StartTime: datetime, EndTime: datetime, EventId: int, State: string, EventType: string, Source: string)

Cree la asignación de tabla Storms_CSV_Mapping correspondiente para los datos ingeridos con el siguiente comando:

.create table Storms ingestion csv mapping 'Storms_CSV_Mapping' '[{"Name":"StartTime","datatype":"datetime","Ordinal":0}, {"Name":"EndTime","datatype":"datetime","Ordinal":1},{"Name":"EventId","datatype":"int","Ordinal":2},{"Name":"State","datatype":"string","Ordinal":3},{"Name":"EventType","datatype":"string","Ordinal":4},{"Name":"Source","datatype":"string","Ordinal":5}]'

Cree una directiva de procesamiento por lotes de ingesta en la tabla para la latencia de ingesta en cola configurable.

Sugerencia

La directiva de procesamiento por lotes de ingesta es un optimizador de rendimiento e incluye tres parámetros. Si se cumple la primera condición, se desencadena la ingesta en la tabla de Azure Data Explorer.
```
.alter table Storms policy ingestionbatching @'{"MaximumBatchingTimeSpan":"00:00:15", "MaximumNumberOfItems": 100, "MaximumRawDataSizeMB": 300}'
```
Use la entidad de servicio de Creación de una entidad de servicio de Microsoft Entra para conceder permiso para trabajar con la base de datos.
```
.add database YOUR_DATABASE_NAME admins  ('aadapp=YOUR_APP_ID;YOUR_TENANT_ID') 'AAD App'
```

Ejecución del laboratorio

El siguiente laboratorio está diseñado para ofrecerle la experiencia de empezar a crear datos, configurar el conector de Kafka y transmitir estos datos a Azure Data Explorer con el conector. Después, puede examinar los datos ingeridos.

Clonación del repositorio Git

Clone el repositorio Git del laboratorio.

Cree un directorio local en el equipo.

mkdir ~/kafka-kusto-hol
cd ~/kafka-kusto-hol

Clone el repositorio.

cd ~/kafka-kusto-hol
git clone https://github.com/Azure/azure-kusto-labs
cd azure-kusto-labs/kafka-integration/dockerized-quickstart

Contenido del repositorio clonado

Ejecute el siguiente comando para enumerar el contenido del repositorio clonado:

cd ~/kafka-kusto-hol/azure-kusto-labs/kafka-integration/dockerized-quickstart
tree

El resultado de esta búsqueda es:

├── README.md
├── adx-query.png
├── adx-sink-config.json
├── connector
│   └── Dockerfile
├── docker-compose.yaml
└── storm-events-producer
    ├── Dockerfile
    ├── StormEvents.csv
    ├── go.mod
    ├── go.sum
    ├── kafka
    │   └── kafka.go
    └── main.go

Revisión de los archivos del repositorio clonado

En las secciones siguientes se explican las partes importantes de los archivos del árbol de archivos anterior.

adx-sink-config.json

Este archivo contiene el archivo de propiedades del receptor de Kusto en el que se van a actualizar los detalles de configuración específicos:

{
    "name": "storm",
    "config": {
        "connector.class": "com.microsoft.azure.kusto.kafka.connect.sink.KustoSinkConnector",
        "flush.size.bytes": 10000,
        "flush.interval.ms": 10000,
        "tasks.max": 1,
        "topics": "storm-events",
        "kusto.tables.topics.mapping": "[{'topic': 'storm-events','db': '<enter database name>', 'table': 'Storms','format': 'csv', 'mapping':'Storms_CSV_Mapping'}]",
        "aad.auth.authority": "<enter tenant ID>",
        "aad.auth.appid": "<enter application ID>",
        "aad.auth.appkey": "<enter client secret>",
        "kusto.ingestion.url": "https://ingest-<name of cluster>.<region>.kusto.windows.net",
        "kusto.query.url": "https://<name of cluster>.<region>.kusto.windows.net",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.kafka.connect.storage.StringConverter"
    }
}

Reemplace los valores de los siguientes atributos según su configuración de Azure Data Explorer: aad.auth.authority, aad.auth.appid, aad.auth.appkey, kusto.tables.topics.mapping (nombre de la base de datos), kusto.ingestion.url y kusto.query.url.

Connector/Dockerfile

Este archivo tiene los comandos para generar la imagen de Docker para la instancia del conector. Incluye la descarga del conector desde el directorio de versión del repositorio Git.

Directorio Storm-events-producer

Este directorio tiene un programa de Go que lee un archivo "StormEvents.csv" local y publica los datos en un tema de Kafka.

docker-compose.yaml

version: "2"
services:
  zookeeper:
    image: debezium/zookeeper:1.2
    ports:
      - 2181:2181
  kafka:
    image: debezium/kafka:1.2
    ports:
      - 9092:9092
    links:
      - zookeeper
    depends_on:
      - zookeeper
    environment:
      - ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
  kusto-connect:
    build:
      context: ./connector
      args:
        KUSTO_KAFKA_SINK_VERSION: 1.0.1
    ports:
      - 8083:8083
    links:
      - kafka
    depends_on:
      - kafka
    environment:
      - BOOTSTRAP_SERVERS=kafka:9092
      - GROUP_ID=adx
      - CONFIG_STORAGE_TOPIC=my_connect_configs
      - OFFSET_STORAGE_TOPIC=my_connect_offsets
      - STATUS_STORAGE_TOPIC=my_connect_statuses
  events-producer:
    build:
      context: ./storm-events-producer
    links:
      - kafka
    depends_on:
      - kafka
    environment:
      - KAFKA_BOOTSTRAP_SERVER=kafka:9092
      - KAFKA_TOPIC=storm-events
      - SOURCE_FILE=StormEvents.csv

Inicio de los contenedores

Inicie los contenedores en un terminal:

docker-compose up

La aplicación del productor comenzará a enviar eventos al tema storm-events. Debería ver registros similares a los siguientes:

....
events-producer_1  | sent message to partition 0 offset 0
events-producer_1  | event  2007-01-01 00:00:00.0000000,2007-01-01 00:00:00.0000000,13208,NORTH CAROLINA,Thunderstorm Wind,Public
events-producer_1  |
events-producer_1  | sent message to partition 0 offset 1
events-producer_1  | event  2007-01-01 00:00:00.0000000,2007-01-01 05:00:00.0000000,23358,WISCONSIN,Winter Storm,COOP Observer
....

Para comprobar los registros, ejecute el siguiente comando en un terminal independiente:
```
docker-compose logs -f | grep kusto-connect
```

Inicio del conector

Use una llamada REST de Kafka Connect para iniciar el conector.

En un terminal independiente, inicie la tarea del receptor con el siguiente comando:

curl -X POST -H "Content-Type: application/json" --data @adx-sink-config.json http://localhost:8083/connectors

Para comprobar el estado, ejecute el siguiente comando en un terminal independiente:
```
curl http://localhost:8083/connectors/storm/status
```

El conector comenzará a poner en cola los procesos de ingesta en Azure Data Explorer.

Nota

Si tiene problemas con el conector de registro, cree una incidencia.

Consulta y revisión de los datos

Confirmación de la ingesta de datos

Espere a que lleguen los datos a la tabla Storms. Para confirmar la transferencia de datos, compruebe el número de filas:
```
Storms | count
```
Confirme que no hay errores en el proceso de ingesta:
```
.show ingestion failures
```
Una vez que aparezcan los datos, pruebe a realizar algunas consultas.

Consultar los datos

Para ver todos los registros, ejecute la siguiente consulta:
```
Storms
```

Use where y project para filtrar datos específicos:

Storms
| where EventType == 'Drought' and State == 'TEXAS'
| project StartTime, EndTime, Source, EventId

Use el operador summarize:

Storms
| summarize event_count=count() by State
| where event_count > 10
| project State, event_count
| render columnchart

Captura de pantalla del gráfico de columnas de consulta de Kafka da como resultado Azure Data Explorer.

Para obtener más ejemplos de consultas e instrucciones, consulte Escritura de consultas en KQL y documentación de Lenguaje de consulta Kusto.

Reset

Para restablecer, realice los pasos siguientes:

Detener los contenedores (docker-compose down -v)
Eliminar (drop table Storms)
Volver a crear la tabla Storms
Volver a crear la asignación de tabla
Reiniciar los contenedores (docker-compose up)

Limpieza de recursos

Para eliminar los recursos de Azure Data Explorer, use az cluster delete o az Kusto database delete:

az kusto cluster delete -n <cluster name> -g <resource group name>
az kusto database delete -n <database name> --cluster-name <cluster name> -g <resource group name>

Optimización del conector receptor de Kafka

Ajuste el conector receptor de Kafka para que funcione con la directiva de procesamiento por lotes de ingesta:

Ajuste el límite de tamaño flush.size.bytes del receptor de Kafka a partir de 1 MB, aumentando en incrementos de 10 MB o 100 MB.
Cuando se usa el receptor de Kafka, los datos se agregan dos veces. En el lado del conector, los datos se agregan según la configuración de vaciado y en el del servicio Azure Data Explorer según la directiva de procesamiento por lotes. Si el tiempo de procesamiento por lotes es demasiado corto y el conector y el servicio no pueden ingerir datos, se debe aumentar el tiempo de procesamiento por lotes. Establezca el tamaño del procesamiento por lotes en 1 GB y aumente o disminuya en incrementos de 100 MB según sea necesario. Por ejemplo, si el tamaño de vaciado es de 1 MB y el tamaño de la directiva de procesamiento por lotes es de 100 MB, después de que el conector receptor de Kafka agregue un lote de 100 MB, el servicio Azure Data Explorer ingerirá un lote de 100 MB. Si el tiempo de la directiva de procesamiento por lotes es de 20 segundos y el conector receptor de Kafka vacía 50 MB en un período de 20 segundos, el servicio ingerirá un lote de 50 MB.
Puede escalar agregando instancias y particiones de Kafka. Aumente tasks.max hasta alcanzar el número de particiones. Cree una partición si tiene suficientes datos para generar un blob del tamaño de la opción flush.size.bytes. Si el blob es más pequeño, el lote se procesa cuando alcanza el límite de tiempo, por lo que la partición no recibirá suficiente rendimiento. Un gran número de particiones significa una mayor sobrecarga de procesamiento.

Más información sobre Arquitecturas de macrodatos.
Obtenga información sobre la Ingesta de datos de ejemplo con formato JSON en Azure Data Explorer.
Para laboratorios de Kafka adicionales:

Ingesta de datos de Apache Kafka en Azure Data Explorer

Requisitos previos

Creación de una entidad de servicio de Microsoft Entra

Creación de una tabla de destino

Ejecución del laboratorio

Clonación del repositorio Git

Contenido del repositorio clonado

Revisión de los archivos del repositorio clonado

adx-sink-config.json

Connector/Dockerfile

Directorio Storm-events-producer

docker-compose.yaml

Inicio de los contenedores

Inicio del conector

Consulta y revisión de los datos

Confirmación de la ingesta de datos

Consultar los datos

Reset

Limpieza de recursos

Optimización del conector receptor de Kafka

Comentarios

Comentarios

Recursos adicionales

Ingesta de datos de Apache Kafka en Azure Data Explorer

Requisitos previos

Creación de una entidad de servicio de Microsoft Entra

Creación de una tabla de destino

Ejecución del laboratorio

Clonación del repositorio Git

Contenido del repositorio clonado

Revisión de los archivos del repositorio clonado

adx-sink-config.json

Connector/Dockerfile

Directorio Storm-events-producer

docker-compose.yaml

Inicio de los contenedores

Inicio del conector

Consulta y revisión de los datos

Confirmación de la ingesta de datos

Consultar los datos

Reset

Limpieza de recursos

Optimización del conector receptor de Kafka

Contenido relacionado

Comentarios

Comentarios

Recursos adicionales