Écrire des données de Kafka vers BigQuery à l'aide de Dataflow

Cette page explique comment utiliser Dataflow pour lire des données à partir de Google Cloud Managed Service pour Apache Kafka et écrire les enregistrements dans une table BigQuery. Ce tutoriel utilise le modèle Apache Kafka vers BigQuery pour créer la tâche Dataflow.

Présentation

Apache Kafka est une plate-forme Open Source pour les événements de traitement en flux continu. Kafka est couramment utilisé dans les architectures distribuées pour permettre la communication entre des composants faiblement couplés. Vous pouvez utiliser Dataflow pour lire des événements à partir de Kafka, les traiter et écrire les résultats dans une table BigQuery pour une analyse plus approfondie.

Managed Service pour Apache Kafka est un Google Cloud service qui vous aide à exécuter des clusters Kafka sécurisés et évolutifs.

Architecture axée sur les événements à l'aide d'Apache Kafka

Autorisations requises

Le compte de service du nœud de calcul Dataflow doit disposer des rôles IAM (Identity and Access Management) suivants :

• Client Managed Kafka (roles/managedkafka.client)
• Éditeur de données BigQuery (roles/bigquery.dataEditor)

Pour en savoir plus, consultez la section Sécurité et autorisations pour Dataflow.

Créer un cluster Kafka

Au cours de cette étape, vous allez créer un cluster Managed Service pour Apache Kafka. Pour en savoir plus, consultez la section Créer un cluster Managed Service pour Apache Kafka.

Console

1. Accédez à la page Managed Service pour Apache Kafka > Clusters.

   Accéder aux clusters

2. Cliquez sur Créer.

3. Dans la zone Nom du cluster, saisissez un nom pour le cluster.

4. Dans la liste Région, sélectionnez un emplacement pour le cluster.

5. Cliquez sur Créer.

gcloud

Exécutez la managed-kafka clusters create commande.

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

Remplacez les éléments suivants :

• CLUSTER : nom du cluster
• REGION : région dans laquelle vous avez créé le sous-réseau
• PROJECT_ID : ID de votre projet
• SUBNET_NAME: sous-réseau dans lequel vous souhaitez déployer le cluster

La création d'un cluster prend généralement entre 20 et 30 minutes.

Créer un sujet Kafka

Une fois le cluster Managed Service pour Apache Kafka créé, créez un sujet.

Console

1. Accédez à la page Managed Service pour Apache Kafka > Clusters.

   Accéder aux clusters

2. Cliquez sur le nom du cluster.

3. Sur la page d'informations du cluster, cliquez sur Créer un sujet.

4. Dans la zone Nom du sujet, saisissez un nom pour le sujet.

5. Cliquez sur Créer.

gcloud

Exécutez la managed-kafka topics create commande.

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

Remplacez les éléments suivants :

• TOPIC_NAME : nom du sujet à créer

Créer une table BigQuery

Au cours de cette étape, vous allez créer une table BigQuery avec le schéma suivant :

Nom de la colonne	Type de données
name	STRING
customer_id	INTEGER

Si vous ne disposez pas encore d'un ensemble de données BigQuery, commencez par en créer un. Pour en savoir plus, consultez la section Créer des ensembles de données. Créez ensuite une table vide :

Console

1. Accédez à la page BigQuery.

   Accéder à BigQuery

2. Dans le volet Explorateur, développez votre projet, puis sélectionnez un ensemble de données.

3. Dans la section d'informations Ensemble de données, cliquez sur Créer une table.

4. Dans la liste Créer une table à partir de, sélectionnez Table vide.

5. Dans la zone Table, saisissez le nom de la table.

6. Dans la section Schéma, cliquez sur Modifier sous forme de texte.

7. Collez la définition de schéma suivante :

   name:STRING,
   customer_id:INTEGER
   

8. Cliquez sur Créer une table.

gcloud

Exécutez la bq mk commande.

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

Remplacez les éléments suivants :

• PROJECT_ID : ID de votre projet
• DATASET_NAME : nom de l'ensemble de données
• TABLE_NAME : nom de la table à créer

Exécuter la tâche Dataflow

Après avoir créé le cluster Kafka et la table BigQuery, exécutez le modèle Dataflow.

Remarque : Vous pouvez également utiliser ce modèle pour lire des données à partir d'un cluster Kafka externe. Vous devrez peut-être configurer la mise en réseau afin que les VM de nœud de calcul Dataflow puissent atteindre les agents de votre cluster. Pour en savoir plus, consultez la section Configurer la mise en réseau.

Console

Commencez par obtenir l'adresse du serveur d'amorçage du cluster :

1. Dans la Google Cloud console, accédez à la page Clusters.

   Accéder aux clusters

2. Cliquez sur le nom du cluster.

3. Cliquez sur l'onglet Configurations.

4. Copiez l'adresse du serveur d'amorçage à partir de URL d'amorçage.

Exécutez ensuite le modèle pour créer la tâche Dataflow :

1. Accédez à la page Dataflow > Tâches.

   Accéder aux tâches

2. Cliquez sur Créer une tâche à partir d'un modèle.

3. Dans le champ Nom de la tâche, saisissez kafka-to-bq.

4. Pour Point de terminaison régional, sélectionnez la région dans laquelle se trouve votre cluster Managed Service pour Apache Kafka.

5. Sélectionnez le modèle "Kafka vers BigQuery".

6. Saisissez les paramètres de modèle suivants :

   • Serveur d'amorçage Kafka : adresse du serveur d'amorçage
   • Sujet Kafka source : nom du sujet à lire
   • Mode d'authentification de la source Kafka : APPLICATION_DEFAULT_CREDENTIALS
   • Format de message Kafka : JSON
   • Stratégie de nom de table : SINGLE_TABLE_NAME
   • Table BigQuery de sortie : table BigQuery, mise en forme comme suit : PROJECT_ID:DATASET_NAME.TABLE_NAME

7. Sous File d'attente de lettres mortes, cochez Écrire les erreurs dans BigQuery.

8. Saisissez un nom de table BigQuery pour la file d'attente de lettres mortes, mis en forme comme suit : PROJECT_ID:DATASET_NAME.ERROR_TABLE_NAME

   Ne créez pas cette table à l'avance. Le pipeline la crée.

9. Cliquez sur Exécuter la tâche.

gcloud

Exécutez la dataflow flex-template run commande.

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

Remplacez les variables suivantes :

• LOCATION: région dans laquelle se trouve votre Managed Service pour Apache Kafka
• PROJECT_ID : nom de votre Google Cloud projet
• CLUSTER_ID : nom du cluster
• TOPIC : nom du sujet Kafka
• DATASET_NAME : nom de l'ensemble de données
• TABLE_NAME : nom de la table
• ERROR_TABLE_NAME: nom de table BigQuery pour la file d'attente de lettres mortes

Ne créez pas la table pour la file d'attente de lettres mortes à l'avance. Le pipeline la crée.

Envoyer des messages à Kafka

Une fois la tâche Dataflow démarrée, vous pouvez envoyer des messages à Kafka, et le pipeline les écrit dans BigQuery.

1. Créez une VM dans le même sous-réseau que le cluster Kafka et installez les outils de ligne de commande Kafka. Pour obtenir des instructions détaillées, consultez la section Configurer une machine cliente dans Publier et consommer des messages avec la CLI.

2. Exécutez la commande suivante pour écrire des messages dans le sujet Kafka :

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

   Remplacez les variables suivantes :

   • TOPIC : nom du sujet Kafka
   • CLUSTER_ID : nom du cluster
   • LOCATION : région dans laquelle se trouve votre cluster
   • PROJECT_ID : nom de votre Google Cloud projet

3. À l'invite, saisissez les lignes de texte suivantes pour envoyer des messages à Kafka :

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

Utiliser une file d'attente de lettres mortes

Pendant l'exécution de la tâche, le pipeline peut ne pas parvenir à écrire des messages individuels dans BigQuery. Voici les erreurs possibles :

• Erreurs de sérialisation, y compris un format JSON mal formaté.
• Erreurs de conversion de type, causées par une incohérence dans le schéma de la table et les données JSON.
• Champs supplémentaires dans les données JSON qui ne sont pas présents dans le schéma de la table.

Ces erreurs n'entraînent pas l'échec de la tâche et n'apparaissent pas comme des erreurs dans le journal de tâche Dataflow. Au lieu de cela, le pipeline utilise une file d'attente de lettres mortes pour gérer ces types d'erreurs.

Pour activer la file d'attente de lettres mortes lorsque vous exécutez le modèle, définissez les paramètres de modèle suivants :

• useBigQueryDLQ: true
• outputDeadletterTable: nom de table BigQuery complet, par exemple my-project:dataset1.errors

Le pipeline crée automatiquement la table. Si une erreur se produit lors du traitement d'un message Kafka, le pipeline écrit une entrée d'erreur dans la table.

Exemples de messages d'erreur :

Type d'erreur	Données d'événement	errorMessage
Erreur de sérialisation	"Hello world"	Échec de la sérialisation de JSON vers la ligne de table : "Hello world"
Erreur de conversion du type	{"name":"Emily","customer_id":"abc"}	{ "errors" : [ { "debugInfo" : "", "location" : "age", "message" : "Cannot convert value to integer (bad value): abc", "reason" : "invalid" } ], "index" : 0 }
Champ inconnu	{"name":"Zoe","age":34}	{ "errors" : [ { "debugInfo" : "", "location" : "age", "message" : "no such field: customer_id.", "reason" : "invalid" } ], "index" : 0 }

Utiliser des types de données BigQuery

En interne, le connecteur d'E/S Kafka convertit les charges utiles des messages JSON en objets TableRow Apache Beam et traduit les valeurs de champ TableRow en types BigQuery.

Le tableau suivant présente les représentations JSON des types de données BigQuery .

Type BigQuery	Représentation JSON
ARRAY	[1.2,3]
BOOL	true
DATE	"2022-07-01"
DATETIME	"2022-07-01 12:00:00.00"
DECIMAL	5.2E11
FLOAT64	3.142
GEOGRAPHY	"POINT(1 2)" Spécifiez la zone géographique à l'aide de texte connu (WKT) ou de GeoJSON, mis en forme en tant que chaîne. Pour en savoir plus, consultez la section Charger des données géospatiales.
INT64	10
INTERVAL	"0-13 370 48:61:61"
STRING	"string_val"
TIMESTAMP	"2022-07-01T12:00:00.00Z" Utilisez la méthode JavaScript Date.toJSON pour mettre en forme la valeur.

Données structurées

Si vos messages JSON suivent un schéma cohérent, vous pouvez représenter des objets JSON à l'aide du STRUCT type de données dans BigQuery.

Dans l'exemple suivant, le champ answers est un objet JSON avec deux sous-champs, a et b :

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

L'instruction SQL suivante crée une table BigQuery avec un schéma compatible :

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

La table obtenue ressemble à ceci :

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

Données semi-structurées

Si vos messages JSON ne suivent pas un schéma strict, envisagez de les stocker dans BigQuery en tant que JSON type de données. En stockant des données JSON en tant que type JSON, vous n'avez pas besoin de définir le schéma à l'avance. Après l'ingestion des données, vous pouvez interroger les données à l'aide de l'accès aux champs (notation par points) et des opérateurs d'accès aux tableaux dans GoogleSQL. Pour en savoir plus, consultez la page Utiliser des données JSON en langage GoogleSQL.

Utiliser une UDF pour transformer les données

Ce tutoriel suppose que les messages Kafka sont mis en forme au format JSON et que le schéma de la table BigQuery correspond aux données JSON, sans qu'aucune transformation ne soit appliquée aux données.

Vous pouvez éventuellement fournir une fonction définie par l'utilisateur (UDF) en JavaScript qui transforme les données avant leur écriture dans BigQuery. La fonction peut également effectuer un traitement supplémentaire, tel que le filtrage, la suppression des informations personnelles ou l'enrichissement des données à l'aide de champs supplémentaires.

Pour en savoir plus, consultez la page Créer des fonctions définies par l'utilisateur pour les modèles Dataflow.

Étape suivante

• Utiliser Dataflow avec Managed Service pour Apache Kafka
• En savoir plus sur BigQuery