Escribe datos de Kafka en BigQuery con Dataflow

En esta página, se muestra cómo usar Dataflow para leer datos de Google Cloud Managed Service para Apache Kafka y escribir los registros en una tabla de BigQuery. En este instructivo, se usa la plantilla de Apache Kafka a BigQuery para crear el trabajo de Dataflow.

Descripción general

Apache Kafka es una plataforma de código abierto para eventos de transmisión. Kafka se suele usar en arquitecturas distribuidas para permitir la comunicación entre componentes con acoplamiento bajo. Puedes usar Dataflow para leer eventos de Kafka, procesarlos y escribir los resultados en una tabla de BigQuery para su análisis posterior.

Managed Service para Apache Kafka es un Google Cloud servicio que te ayuda a ejecutar clústeres de Kafka seguros y escalables.

Permisos necesarios

La cuenta de servicio del trabajador de Dataflow debe tener los siguientes roles de Identity and Access Management (IAM):

• Cliente de Kafka administrado (roles/managedkafka.client)
• Editor de datos de BigQuery (roles/bigquery.dataEditor)

Para obtener más información, consulta Seguridad y permisos de Dataflow.

Crear clúster de Kafka

En este paso, crearás un clúster de Managed Service para Apache Kafka. Para obtener más información, consulta Crea un clúster de Managed Service para Apache Kafka.

gcloud managed-kafka clusters create CLUSTER \
--location=REGION \
--cpu=3 \
--memory=3GiB \
--subnets=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME

Por lo general, la creación de un clúster tarda entre 20 y 30 minutos.

Crear tema de Kafka

Después de crear el clúster de Managed Service para Apache Kafka, crea un tema.

gcloud managed-kafka topics create TOPIC_NAME \
--cluster=CLUSTER \
--location=REGION \
--partitions=10 \
--replication-factor=3

Crea una tabla de BigQuery

En este paso, crearás una tabla de BigQuery con el siguiente esquema:

Si aún no tienes un conjunto de datos de BigQuery, primero crea uno. Para obtener más información, consulta Crea conjuntos de datos. Luego, crea una tabla vacía nueva:

bq mk --table \
  PROJECT_ID:DATASET_NAME.TABLE_NAME \
  name:STRING,customer_id:INTEGER

Ejecuta el trabajo de Dataflow:

Después de crear el clúster de Kafka y la tabla de BigQuery, ejecuta la plantilla de Dataflow.

gcloud dataflow flex-template run kafka-to-bq \
--template-file-gcs-location gs://dataflow-templates/latest/flex/Kafka_to_BigQuery \
--region LOCATION \
--parameters \
readBootstrapServerAndTopic=projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC,\
persistKafkaKey=false,\
writeMode=SINGLE_TABLE_NAME,\
kafkaReadOffset=earliest,\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
outputTableSpec=PROJECT_ID:DATASET_NAME.TABLE_NAME\
useBigQueryDLQ=true,\
outputDeadletterTable=PROJECT_ID:DATASET_NAME.ERROR_TABLE_NAME

Envía mensajes a Kafka

Después de que se inicia el trabajo de Dataflow, puedes enviar mensajes a Kafka y la canalización los escribe en BigQuery.

1. Crea una VM en la misma subred que el clúster de Kafka y, luego, instala las herramientas de línea de comandos de Kafka. Para obtener instrucciones detalladas, consulta Configura una máquina cliente en Publica y consume mensajes con la CLI.

2. Ejecuta el siguiente comando para escribir mensajes en el tema de Kafka:

   kafka-console-producer.sh \
    --topic TOPIC \
    --bootstrap-server bootstrap.CLUSTER_ID.LOCATION.managedkafka.PROJECT_ID.cloud.goog:9092 \
    --producer.config client.properties

   Reemplaza las siguientes variables:

   • TOPIC: Es el nombre del tema de Kafka.
   • CLUSTER_ID: Es el nombre del clúster.
   • LOCATION: Es la región en la que se encuentra el clúster.
   • PROJECT_ID: Es el nombre de tu Google Cloud proyecto de.

3. En la solicitud, ingresa las siguientes líneas de texto para enviar mensajes a Kafka:

   {"name": "Alice", "customer_id": 1}
   {"name": "Bob", "customer_id": 2}
   {"name": "Charles", "customer_id": 3}
   

Usa una cola de mensajes no entregados

Mientras se ejecuta el trabajo, es posible que la canalización no pueda escribir mensajes individuales en BigQuery. Los errores posibles incluyen los siguientes:

• Errores de serialización, incluido el JSON con formato incorrecto
• Errores de conversión de tipos, causados por una falta de coincidencia en el esquema de la tabla y los datos JSON
• Campos adicionales en los datos JSON que no están presentes en el esquema de la tabla

Estos errores no hacen que falle el trabajo y no aparecen como errores en el registro de trabajos de Dataflow. En su lugar, la canalización usa una cola de mensajes no entregados para controlar estos tipos de errores.

Para habilitar la cola de mensajes no entregados cuando ejecutas la plantilla, establece los siguientes parámetros de plantilla:

• useBigQueryDLQ: true
• outputDeadletterTable: Es un nombre de tabla de BigQuery completamente calificado; por ejemplo, my-project:dataset1.errors.

La canalización crea la tabla automáticamente. Si se produce un error cuando se procesa un mensaje de Kafka, la canalización escribe una entrada de error en la tabla.

Ejemplos de mensajes de error:

Trabaja con tipos de datos de BigQuery

De forma interna, el conector de E/S de Kafka convierte las cargas útiles de mensajes JSON en objetos TableRow de Apache Beam y traduce los valores de los campos TableRow en tipos de BigQuery.

En la siguiente tabla, se muestran las representaciones JSON de los tipos de datos de BigQuery .

Datos estructurados

Si tus mensajes JSON siguen un esquema coherente, puedes representar objetos JSON con el STRUCT tipo de datos en BigQuery.

En el siguiente ejemplo, el campo answers es un objeto JSON con dos subcampos, a y b:

{"name":"Emily","answers":{"a":"yes","b":"no"}}

La siguiente instrucción de SQL crea una tabla de BigQuery con un esquema compatible:

CREATE TABLE my_dataset.kafka_events (name STRING, answers STRUCT<a STRING, b STRING>);

La tabla resultante se ve de la siguiente manera:

+-------+----------------------+
| name  |       answers        |
+-------+----------------------+
| Emily | {"a":"yes","b":"no"} |
+-------+----------------------+

Datos semiestructurados

Si tus mensajes JSON no siguen un esquema estricto, considera almacenarlos en BigQuery como un JSON tipo de datos. Cuando almacenas datos JSON como un tipo JSON, no necesitas definir el esquema por adelantado. Después de la transferencia de datos, puedes consultar los datos con los operadores de acceso a campos (notación de puntos) y acceso a arrays en GoogleSQL. Para obtener más información, consulta Trabaja con datos JSON en GoogleSQL.

Usa una UDF para transformar los datos

En este instructivo, se supone que los mensajes de Kafka tienen formato JSON y que el esquema de la tabla de BigQuery coincide con los datos JSON, sin que se apliquen transformaciones a los datos.

De forma opcional, puedes proporcionar una función definida por el usuario (UDF) de JavaScript que transforme los datos antes de que se escriban en BigQuery. La UDF también puede realizar un procesamiento adicional, como filtrar, quitar información de identificación personal (PII) o enriquecer los datos con campos adicionales.

Para obtener más información, consulta Crea funciones definidas por el usuario para plantillas de Dataflow.

¿Qué sigue?

• Usa Dataflow con Managed Service para Apache Kafka
• Más información de BigQuery