Le modèle Apache Kafka vers Cloud Storage est un pipeline de flux de données qui ingère les données textuelles de Google Cloud Managed Service pour Apache Kafka et génère les enregistrements dans Cloud Storage.
Vous pouvez également utiliser le modèle Apache Kafka vers BigQuery avec Kafka autogéré ou externe.
Conditions requises pour ce pipeline
- Le bucket Cloud Storage de sortie doit exister.
- Le serveur de courtiers Apache Kafka doit être en cours d'exécution et joignable depuis les machines de nœud de calcul Dataflow.
- Les sujets Apache Kafka doivent exister.
Format de message Kafka
Le modèle Apache Kafka vers Cloud Storage permet de lire des messages de Kafka aux formats suivants : CONFLUENT_AVRO_WIRE_FORMAT
et JSON
.
Format du fichier de sortie
Le format du fichier de sortie est identique à celui du message Kafka d'entrée. Par exemple, si vous sélectionnez JSON pour le format de message Kafka, les fichiers JSON sont écrits dans le bucket Cloud Storage de sortie.
Authentification
Le modèle Apache Kafka vers Cloud Storage est compatible avec l'authentification SASL/PLAIN pour les courtiers Kafka.
Paramètres de modèle
Paramètres obligatoires
- readBootstrapServerAndTopic : Sujet Kafka à partir duquel lire l'entrée.
- kafkaReadAuthenticationMode : mode d'authentification à utiliser avec le cluster Kafka. Utilisez "NONE" pour désactiver l'authentification et "SASL_PLAIN" pour le nom d'utilisateur et le mot de passe SASL/PLAIN. Apache Kafka pour BigQuery n'est compatible qu'avec le mode d'authentification SASL_PLAIN. Valeur par défaut : SASL_PLAIN.
- outputDirectory : Chemin d'accès et préfixe du nom de fichier pour l'écriture des fichiers de sortie. Doit se terminer par une barre oblique. (par exemple : gs://votre-bucket/votre-chemin-d'accès).
- messageFormat : format des messages Kafka à lire. Les valeurs acceptées sont AVRO_CONFLUENT_WIRE_FORMAT (Avro encodé par Confluent Schema Registry), AVRO_BINARY_ENCODING (Avro binaire simple) et JSON. Valeur par défaut : AVRO_CONFLUENT_WIRE_FORMAT.
Paramètres facultatifs
- windowDuration : durée/taille de la fenêtre dans laquelle les données seront écrites dans Cloud Storage. Les formats autorisés sont les suivants : Ns (pour les secondes, exemple : 5s), Nm (pour les minutes, exemple : 12m), Nh (pour les heures, exemple : 2h). (exemple : 5 m). La valeur par défaut est "5m".
- outputFilenamePrefix : préfixe à placer sur chaque fichier ciblé sur une fenêtre. (par exemple : output-). La valeur par défaut est "output".
- numShards: Nombre maximal de partitions de sortie générées lors de l'écriture. Un nombre plus élevé de segments entraîne un débit plus élevé pour l'écriture dans Cloud Storage, mais potentiellement un coût d'agrégation de données plus élevé entre les partitions lors du traitement des fichiers Cloud Storage de sortie. La valeur par défaut est déterminée par Dataflow.
- enableCommitOffsets : commit des décalages des messages traités vers Kafka. Si cette option est activée, les écarts ou le traitement en double des messages seront minimisés lors du redémarrage du pipeline. L'ID du groupe de consommateurs doit être spécifié. La valeur par défaut est "false".
- consumerGroupId : identifiant unique du groupe de consommateurs auquel ce pipeline appartient. Obligatoire si l'option "Commit Offsets to Kafka" (Enregistrer les décalages dans Kafka) est activée. La valeur par défaut est vide.
- kafkaReadOffset : point de départ de la lecture des messages lorsqu'il n'existe aucun décalage validé. Le premier commence au début, le dernier à partir du message le plus récent. Valeur par défaut : le plus récent.
- kafkaReadUsernameSecretId : ID du secret Google Cloud Secret Manager contenant le nom d'utilisateur Kafka à utiliser avec l'authentification SASL_PLAIN. (Exemple : projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>). La valeur par défaut est vide.
- kafkaReadPasswordSecretId : ID du secret Secret Manager de Google Cloud contenant le mot de passe Kafka à utiliser avec l'authentification SASL_PLAIN. (Exemple : projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>). La valeur par défaut est vide.
- schemaFormat : format de schéma de Kafka. Peut être fourni sous forme de SINGLE_SCHEMA_FILE ou SCHEMA_REGISTRY. Si SINGLE_SCHEMA_FILE est spécifié, tous les messages doivent avoir le schéma mentionné dans le fichier de schéma Avro. Si schema_REGISTRY est spécifié, les messages peuvent avoir un seul schéma ou plusieurs. La valeur par défaut est SINGLE_schema_FILE.
- confluentAvroSchemaPath : chemin d'accès Google Cloud Storage au fichier de schéma Avro unique utilisé pour décoder tous les messages d'un sujet. La valeur par défaut est vide.
- schemaRegistryConnectionUrl : URL de l'instance Confluent Schema Registry utilisée pour gérer les schémas Avro pour le décodage des messages. La valeur par défaut est vide.
- binaryAvroSchemaPath : chemin d'accès Google Cloud Storage au fichier de schéma Avro utilisé pour décoder les messages Avro encodés en binaire. La valeur par défaut est vide.
Exécuter le modèle
Console
- Accédez à la page Dataflow Créer un job à partir d'un modèle. Accéder à la page Créer un job à partir d'un modèle
- Dans le champ Nom du job, saisissez un nom de job unique.
- Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. La région par défaut est
us-central1
.Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter un job Dataflow, consultez la page Emplacements Dataflow.
- Dans le menu déroulant Modèle Dataflow, sélectionnez the Kafka to Cloud Storage template.
- Dans les champs fournis, saisissez vos valeurs de paramètres.
- Facultatif : Pour passer du traitement de type "exactement une fois" au mode de traitement par flux de type "au moins une fois", sélectionnez Au moins une fois.
- Cliquez sur Run Job (Exécuter la tâche).
gcloud
Dans le shell ou le terminal, exécutez le modèle :
gcloud dataflow flex-template run JOB_NAME \ --project=PROJECT_ID \ --region=REGION_NAME \ --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Kafka_to_Cloud Storage \ --parameters \ outputTableSpec=BIGQUERY_TABLE,\ inputTopics=KAFKA_TOPICS,\ javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\ javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\ bootstrapServers=KAFKA_SERVER_ADDRESSES
Remplacez les éléments suivants :
PROJECT_ID
: ID du projet Google Cloud dans lequel vous souhaitez exécuter le job DataflowJOB_NAME
: nom de job unique de votre choixREGION_NAME
: région dans laquelle vous souhaitez déployer votre job Dataflow, par exempleus-central1
VERSION
: version du modèle que vous souhaitez utiliserVous pouvez utiliser les valeurs suivantes :
latest
pour utiliser la dernière version du modèle, disponible dans le dossier parent non daté du bucket gs://dataflow-templates-REGION_NAME/latest/- Le nom de la version, par exemple
2023-09-12-00_RC00
, pour utiliser une version spécifique du modèle, qui est imbriqué dans le dossier parent daté respectif dans le bucket : gs://dataflow-templates-REGION_NAME/
BIGQUERY_TABLE
: nom de votre table Cloud Storage.KAFKA_TOPICS
: liste des sujets Apache Kakfa. Si plusieurs sujets sont fournis, vous devez échapper les virgules. Consultezgcloud topic escaping
.PATH_TO_JAVASCRIPT_UDF_FILE
: URI Cloud Storage du fichier.js
contenant la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser (par exemple,gs://my-bucket/my-udfs/my_file.js
).JAVASCRIPT_FUNCTION
: Nom de la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser.Par exemple, si le code de votre fonction JavaScript est
myTransform(inJson) { /*...do stuff...*/ }
, le nom de la fonction estmyTransform
. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.KAFKA_SERVER_ADDRESSES
: liste d'adresses IP du serveur de courtiers Apache Kafka. Chaque adresse IP doit comporter le numéro de port à partir duquel le serveur est accessible. Exemple :35.70.252.199:9092
. Si plusieurs adresses sont fournies, vous devez échapper les virgules. Consultezgcloud topic escaping
.
API
Pour exécuter le modèle à l'aide de l'API REST, envoyez une requête HTTP POST. Pour en savoir plus sur l'API, ses autorisations et leurs champs d'application, consultez la section projects.templates.launch
.
POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch { "launch_parameter": { "jobName": "JOB_NAME", "parameters": { "outputTableSpec": "BIGQUERY_TABLE", "inputTopics": "KAFKA_TOPICS", "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE", "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION", "bootstrapServers": "KAFKA_SERVER_ADDRESSES" }, "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Kafka_to_Cloud Storage", } }
Remplacez les éléments suivants :
PROJECT_ID
: ID du projet Google Cloud dans lequel vous souhaitez exécuter le job DataflowJOB_NAME
: nom de job unique de votre choixLOCATION
: région dans laquelle vous souhaitez déployer votre job Dataflow, par exempleus-central1
VERSION
: version du modèle que vous souhaitez utiliserVous pouvez utiliser les valeurs suivantes :
latest
pour utiliser la dernière version du modèle, disponible dans le dossier parent non daté du bucket gs://dataflow-templates-REGION_NAME/latest/- Le nom de la version, par exemple
2023-09-12-00_RC00
, pour utiliser une version spécifique du modèle, qui est imbriqué dans le dossier parent daté respectif dans le bucket : gs://dataflow-templates-REGION_NAME/
BIGQUERY_TABLE
: nom de votre table Cloud Storage.KAFKA_TOPICS
: liste des sujets Apache Kakfa. Si plusieurs sujets sont fournis, vous devez échapper les virgules. Consultezgcloud topic escaping
.PATH_TO_JAVASCRIPT_UDF_FILE
: URI Cloud Storage du fichier.js
contenant la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser (par exemple,gs://my-bucket/my-udfs/my_file.js
).JAVASCRIPT_FUNCTION
: Nom de la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser.Par exemple, si le code de votre fonction JavaScript est
myTransform(inJson) { /*...do stuff...*/ }
, le nom de la fonction estmyTransform
. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.KAFKA_SERVER_ADDRESSES
: liste d'adresses IP du serveur de courtiers Apache Kafka. Chaque adresse IP doit comporter le numéro de port à partir duquel le serveur est accessible. Exemple :35.70.252.199:9092
. Si plusieurs adresses sont fournies, vous devez échapper les virgules. Consultezgcloud topic escaping
.
Pour en savoir plus, consultez la page Écrire des données de Kafka vers Cloud Storage avec Dataflow.
Étapes suivantes
- Apprenez-en plus sur les modèles Dataflow.
- Consultez la liste des modèles fournis par Google.