Obtención de datos de Kafka

Artículo
11/19/2024

Apache Kafka es una plataforma de streaming distribuida para la creación de canalizaciones de streaming de datos en tiempo real que mueve los datos de forma confiable entre aplicaciones o sistemas. 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 Kusto actúa como conector de Kafka y no requiere el uso de código. Descargue el archivo jar del conector del receptor desde el repositorio Git o desde Confluent Connector Hub.

En este artículo se muestra cómo ingerir datos con Kafka mediante una configuración de Docker autocontenida para simplificar el clúster de Kafka y la configuración del clúster 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.
Base de datos KQL en Microsoft Fabric.
El identificador URI de ingesta de la base de datos y el de consulta que se van a usar en el archivo JSON de configuración. Para obtener más información, consulte Copia de identificador URI.
Azure CLI.
Docker y Docker Compose.

Creación de una entidad de servicio de Microsoft Entra

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

Esta entidad de servicio es la identidad utilizada por el conector para escribir datos en la tabla de Kusto. Se te conceden 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 servicio. 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 los valores appId, password y tenant para usarlos posteriormente.

{
  "appId": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "displayName": "my-service-principal",
  "name": "my-service-principal",
  "password": "00001111-aaaa-2222-bbbb-3333cccc4444",
  "tenant": "00001111-aaaa-2222-bbbb-3333cccc4444"
}

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

Creación de una tabla de destino

Desde el entorno de consulta, cree una tabla llamada 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 ingesta por lotes en la tabla para una latencia de ingesta configurable en cola.

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.
```
.alter table Storms policy ingestionbatching @'{"MaximumBatchingTimeSpan":"00:00:15", "MaximumNumberOfItems": 100, "MaximumRawDataSizeMB": 300}'
```
Use la entidad de servicio de la sección Creación de una entidad de servicio de Microsoft Entra para conceder permisos 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 ofrecerte la experiencia de empezar a crear datos, configurar el conector de Kafka y transmitir estos datos. 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": "<ingestion URI per prerequisites>",
        "kusto.query.url": "<query URI per prerequisites>",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter": "org.apache.kafka.connect.storage.StringConverter"
    }
}

Reemplaza los valores de los siguientes atributos según su configuración: 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 inicia los procesos de ingesta en cola.

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

Una vez que los datos hayan llegado a la tabla Storms, confirma la transferencia de datos; para ello, comprueba el recuento 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
| take 10
```

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 de los resultados del gráfico de columnas de consulta de Kafka conectados.

Para ver más ejemplos de consultas e instrucciones, consulte cómo escribir consultas en KQL y la documentación del 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

Limpie los elementos creados; para ello, vaya al área de trabajo en la que se crearon.

En el área de trabajo, mueve el puntero sobre la base de datos y selecciona el menú Más [...] > Eliminar.
Seleccione Eliminar. No se pueden recuperar los elementos eliminados.

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 según la directiva de procesamiento por lotes. Si el tiempo de procesamiento por lotes es demasiado corto, el conector y el servicio no pueden ingerir datos y 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, el conector receptor de Kafka agrega datos a un lote de 100 MB. Después, el servicio ingiere ese lote. Si el tiempo de la directiva de procesamiento por lotes es de 20 segundos y el conector del 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 se alcanza el límite de tiempo, por lo que la partición no dispone de suficiente rendimiento. Un gran número de particiones significa una mayor sobrecarga de procesamiento.

Creación de una base de datos KQL

Compartir vía

Obtención de datos de Kafka

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

Recursos adicionales

Compartir vía

Obtención de datos de Kafka

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

Recursos adicionales