Modèles de streaming fournis par Google

Google fournit un ensemble de modèles Dataflow Open Source. Pour obtenir des informations générales sur les modèles, consultez la page Modèles Dataflow. Pour obtenir la liste de tous les modèles fournis par Google, consultez la page Premiers pas avec les modèles fournis par Google.

Ce guide répertorie les modèles de streaming.

Abonnement Pub/Sub vers BigQuery

Le modèle Abonnement Pub/Sub vers BigQuery est un pipeline de streaming qui lit les messages au format JSON d'un abonnement Pub/Sub et les écrit dans une table BigQuery. Vous pouvez utiliser ce modèle comme solution rapide pour déplacer des données Pub/Sub vers BigQuery. Le modèle lit les messages au format JSON de Pub/Sub et les convertit en éléments BigQuery.

Conditions requises pour ce pipeline :

  • Les messages Pub/Sub doivent être au format JSON, comme décrit ici. Par exemple, vous pouvez insérer des messages au format {"k1":"v1", "k2":"v2"} dans une table BigQuery comportant deux colonnes nommées k1 et k2, en utilisant le type de données "chaîne".
  • La table de sortie doit exister avant l'exécution du pipeline. Le schéma de la table doit correspondre aux objets JSON d'entrée.

Paramètres de modèle

Paramètres Description
inputSubscription Abonnement en entrée Pub/Sub à lire, au format projects/<project>/subscriptions/<subscription>.
outputTableSpec Emplacement de la table de sortie BigQuery, au format <my-project>:<my-dataset>.<my-table>
outputDeadletterTable Table BigQuery des messages qui n'ont pas pu atteindre la table de sortie, au format <my-project>:<my-dataset>.<my-table>. Si elle n'existe pas, elle est créée lors de l'exécution du pipeline. Si ce paramètre n'est pas spécifié, OUTPUT_TABLE_SPEC_error_records est utilisé à la place.
javascriptTextTransformGcsPath (Facultatif) URI Cloud Storage du fichier .js contenant la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser. Exemple : gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName (Facultatif) 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.

Exécuter le modèle Abonnement Pub/Sub vers BigQuery

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub Subscription to BigQuery template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/PubSub_Subscription_to_BigQuery \
    --region REGION_NAME \
    --staging-location TEMP_LOCATION \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
outputDeadletterTable=PROJECT_ID:DATASET.TABLE_NAME

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • SUBSCRIPTION_NAME : nom de votre abonnement Pub/Sub
  • DATASET : votre ensemble de données BigQuery.
  • TABLE_NAME : nom de votre table BigQuery.

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/PubSub_Subscription_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputSubscription": "projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME",
       "outputTableSpec": "PROJECT_ID:DATASET.TABLE_NAME"
   },
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": "TEMP_LOCATION",
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
   },
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • SUBSCRIPTION_NAME : nom de votre abonnement Pub/Sub
  • DATASET : votre ensemble de données BigQuery.
  • TABLE_NAME : nom de votre table BigQuery.

Sujet Pub/Sub vers BigQuery

Le modèle Sujet Pub/Sub vers BigQuery est un pipeline de streaming qui lit les messages au format JSON d'un sujet Pub/Sub et les écrit dans une table BigQuery. Vous pouvez utiliser ce modèle comme solution rapide pour déplacer des données Pub/Sub vers BigQuery. Le modèle lit les messages au format JSON de Pub/Sub et les convertit en éléments BigQuery.

Conditions requises pour ce pipeline :

  • Les messages Pub/Sub doivent être au format JSON, comme décrit ici. Par exemple, vous pouvez insérer des messages au format {"k1":"v1", "k2":"v2"} dans une table BigQuery comportant deux colonnes nommées k1 et k2, en utilisant le type de données "chaîne".
  • La table de sortie doit exister avant l'exécution du pipeline. Le schéma de la table doit correspondre aux objets JSON d'entrée.

Paramètres de modèle

Paramètres Description
inputTopic Sujet d'entrée Pub/Sub à lire, au format projects/<project>/topics/<topic>.
outputTableSpec Emplacement de la table de sortie BigQuery, au format <my-project>:<my-dataset>.<my-table>
outputDeadletterTable La table BigQuery des messages n'ayant pas pu atteindre la table de sortie Elle doit être au format <my-project>:<my-dataset>.<my-table>. Si elle n'existe pas, elle est créée lors de l'exécution du pipeline. Si ce paramètre n'est pas spécifié, <outputTableSpec>_error_records est utilisé à la place.
javascriptTextTransformGcsPath (Facultatif) URI Cloud Storage du fichier .js contenant la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser. Exemple : gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName (Facultatif) 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.

Exécuter le modèle Sujet Pub/Sub vers BigQuery

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub Topic to BigQuery template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/PubSub_to_BigQuery \
    --region REGION_NAME \
    --staging-location TEMP_LOCATION \
    --parameters \
inputTopic=projects/PROJECT_ID/topics/TOPIC_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
outputDeadletterTable=PROJECT_ID:DATASET.TABLE_NAME

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • TOPIC_NAME : nom de votre sujet Pub/Sub
  • DATASET : votre ensemble de données BigQuery.
  • TABLE_NAME : nom de votre table BigQuery.

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/PubSub_to_BigQuery
{
   "jobName": "JOB_NAME",
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": TEMP_LOCATION,
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
    },
   "parameters": {
       "inputTopic": "projects/PROJECT_ID/topics/TOPIC_NAME",
       "outputTableSpec": "PROJECT_ID:DATASET.TABLE_NAME"
   }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • TOPIC_NAME : nom de votre sujet Pub/Sub
  • DATASET : votre ensemble de données BigQuery.
  • TABLE_NAME : nom de votre table BigQuery.

Pub/Sub Avro vers BigQuery

Le modèle Pub/Sub Avro vers BigQuery est un pipeline de streaming qui ingère les données Avro d'un abonnement Pub/Sub dans une table BigQuery. Toute erreur survenant lors de l'écriture dans la table BigQuery est traitée dans un sujet Pub/Sub non traité.

Conditions requises pour ce pipeline

  • L'abonnement Pub/Sub d'entrée doit exister.
  • Le fichier de schéma des enregistrements Avro doit exister dans Cloud Storage.
  • Le sujet Pub/Sub non traité doit exister.
  • L'ensemble de données BigQuery de sortie doit exister.

Paramètres de modèle

Paramètres Description
schemaPath Emplacement Cloud Storage du fichier de schéma Avro. Par exemple, gs://path/to/my/schema.avsc.
inputSubscription Abonnement en entrée Pub/Sub à lire. Par exemple, projects/<project>/subscriptions/<subscription>.
outputTopic Sujet Pub/Sub à utiliser pour les enregistrements non traités. Par exemple, projects/<project-id>/topics/<topic-name>.
outputTableSpec Emplacement de la table de sortie BigQuery. Par exemple, <my-project>:<my-dataset>.<my-table>. Selon la propriété createDisposition spécifiée, la table de sortie peut être créée automatiquement à l'aide du schéma Avro fourni par l'utilisateur.
writeDisposition (Facultatif) La propriété WriteDisposition de BigQuery. Par exemple, WRITE_APPEND, WRITE_EMPTY ou WRITE_TRUNCATE. Par défaut, WRITE_APPEND.
createDisposition (Facultatif) La propriété CreateDisposition de BigQuery. Par exemple, CREATE_IF_NEEDED et CREATE_NEVER. Par défaut, CREATE_IF_NEEDED.

Exécuter le modèle Pub/Sub Avro vers BigQuery

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub Avro to BigQuery template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud beta dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/PubSub_Avro_to_BigQuery \
    --parameters \
schemaPath=SCHEMA_PATH,\
inputSubscription=SUBSCRIPTION_NAME,\
outputTableSpec=BIGQUERY_TABLE,\
outputTopic=DEADLETTER_TOPIC
  

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • SCHEMA_PATH : chemin d'accès Cloud Storage au fichier de schéma Avro (par exemple, gs://MyBucket/file.avsc)
  • SUBSCRIPTION_NAME : nom de l'abonnement d'entrée Pub/Sub
  • BIGQUERY_TABLE : nom de la table de sortie BigQuery
  • DEADLETTER_TOPIC : sujet Pub/Sub à utiliser pour la file d'attente non traitée

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/PubSub_Avro_to_BigQuery",
      "parameters": {
          "schemaPath": "SCHEMA_PATH",
          "inputSubscription": "SUBSCRIPTION_NAME",
          "outputTableSpec": "BIGQUERY_TABLE",
          "outputTopic": "DEADLETTER_TOPIC"
      }
   }
}
  

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • SCHEMA_PATH : chemin d'accès Cloud Storage au fichier de schéma Avro (par exemple, gs://MyBucket/file.avsc)
  • SUBSCRIPTION_NAME : nom de l'abonnement d'entrée Pub/Sub
  • BIGQUERY_TABLE : nom de la table de sortie BigQuery
  • DEADLETTER_TOPIC : sujet Pub/Sub à utiliser pour la file d'attente non traitée

Pub/Sub vers Pub/Sub

Le modèle Pub/Sub vers Pub/Sub est un pipeline de streaming qui lit les messages d'un abonnement Pub/Sub et les écrit dans un autre sujet Pub/Sub. Le pipeline accepte également une clé facultative d'attribut de message et une valeur qui peut être utilisée pour filtrer les messages devant être écrits dans le sujet Pub/Sub. Vous pouvez utiliser ce modèle pour copier des messages d'un abonnement Pub/Sub à un autre sujet Pub/Sub avec un filtre de message facultatif.

Conditions requises pour ce pipeline :

  • L'abonnement Pub/Sub source doit exister avant l'exécution.
  • Le sujet Pub/Sub de destination doit exister avant l'exécution.

Paramètres de modèle

Paramètres Description
inputSubscription Abonnement Pub/Sub à partir duquel lire l'entrée. Par exemple, projects/<project-id>/subscriptions/<subscription-name>.
outputTopic Sujet Cloud Pub/Sub dans lequel écrire la sortie. Par exemple, projects/<project-id>/topics/<topic-name>.
filterKey (Facultatif) Filtrez les événements en fonction d'une clé d'attribut. Aucun filtre n'est appliqué si filterKey n'est pas spécifié.
filterValue (Facultatif) Valeur d'attribut de filtre à utiliser dans le cas où un filterKey est fourni. Une valeur filterValue nulle est utilisée par défaut.

Exécuter le modèle Pub/Sub vers Pub/Sub

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub to Pub/Sub template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Cloud_PubSub_to_Cloud_PubSub \
    --region REGION_NAME \
    --staging-location TEMP_LOCATION \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME,\
outputTopic=projects/PROJECT_ID/topics/TOPIC_NAME,\
filterKey=FILTER_KEY,\
filterValue=FILTER_VALUE

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • SUBSCRIPTION_NAME : nom de l'abonnement Pub/Sub
  • TOPIC_NAME : nom du sujet Pub/Sub
  • FILTER_KEY : clé d'attribut utilisée pour le filtrage des événements. Aucun filtre n'est appliqué si aucune clé n'est spécifiée.
  • FILTER_VALUE : valeur d'attribut de filtre à utiliser si une clé de filtre d'événement est fournie. Accepte une chaîne de l'expression régulière Java valide en tant que valeur de filtre d'événement. En cas d'expression régulière, l'expression complète doit correspondre pour que le message soit filtré. Les correspondances partielles (sous-chaîne, par exemple) ne seront pas filtrées. Une valeur de filtre d'événement "null" est utilisée par défaut.

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Cloud_PubSub_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": TEMP_LOCATION,
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
    },
   "parameters": {
       "inputSubscription": "projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME",
       "outputTopic": "projects/PROJECT_ID/topics/TOPIC_NAME",
       "filterKey": "FILTER_KEY",
       "filterValue": "FILTER_VALUE"
   }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • SUBSCRIPTION_NAME : nom de l'abonnement Pub/Sub
  • TOPIC_NAME : nom du sujet Pub/Sub
  • FILTER_KEY : clé d'attribut utilisée pour le filtrage des événements. Aucun filtre n'est appliqué si aucune clé n'est spécifiée.
  • FILTER_VALUE : valeur d'attribut de filtre à utiliser si une clé de filtre d'événement est fournie. Accepte une chaîne de l'expression régulière Java valide en tant que valeur de filtre d'événement. En cas d'expression régulière, l'expression complète doit correspondre pour que le message soit filtré. Les correspondances partielles (sous-chaîne, par exemple) ne seront pas filtrées. Une valeur de filtre d'événement "null" est utilisée par défaut.

Pub/Sub vers Splunk

Le modèle Pub/Sub vers Splunk est un pipeline de streaming qui lit les messages d'un abonnement Pub/Sub et écrit leur charge utile dans Splunk via la solution HEC (HTTP Event Collector) de Splunk. Le cas d'utilisation le plus courant de ce modèle est l'exportation de journaux vers Splunk. Pour découvrir un exemple du workflow sous-jacent, consultez la section Déployer des exportations de journaux prêtes pour la production vers Splunk à l'aide de Dataflow.

Avant d'écrire vers Splunk, vous pouvez également appliquer une fonction JavaScript définie par l'utilisateur vers la charge utile du message. Tous les messages dont le traitement échoue sont transférés vers un sujet Pub/Sub non traité en vue d'opérations de dépannage supplémentaires et d'un nouveau traitement.

Pour ajouter une couche de protection à votre jeton HEC, vous pouvez également transmettre une clé Cloud KMS ainsi que le paramètre de jeton HEC encodé en base64 chiffré avec cette clé. Pour en savoir plus sur le chiffrement des paramètres du jeton HEC, consultez la page sur le point de terminaison du chiffrement de l'API Cloud KMS.

Conditions requises pour ce pipeline :

  • L'abonnement Pub/Sub source doit exister avant l'exécution du pipeline.
  • Le sujet Pub/Sub non traité doit exister avant l'exécution du pipeline.
  • Le point de terminaison Splunk HEC doit être accessible à partir du réseau de nœuds de calcul Dataflow.
  • Le jeton de la solution HEC de Splunk doit être généré et disponible.

Paramètres de modèle

Paramètres Description
inputSubscription Abonnement Pub/Sub à partir duquel lire l'entrée. Par exemple, projects/<project-id>/subscriptions/<subscription-name>.
token Jeton d'authentification HEC Splunk. Cette chaîne encodée en base64 peut être chiffrée avec une clé Cloud KMS pour plus de sécurité.
url URL HEC Splunk. Il doit être routable depuis le VPC dans lequel le pipeline est exécuté. Par exemple, https://splunk-hec-host:8088.
outputDeadletterTopic Sujet Pub/Sub pour transférer les messages non distribuables. Par exemple, projects/<project-id>/topics/<topic-name>.
javascriptTextTransformGcsPath (Facultatif) URI Cloud Storage du fichier .js contenant la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser. Exemple : gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName (Facultatif) 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.
batchCount (Facultatif) Taille de lot pour l'envoi de plusieurs événements vers Splunk. Valeur par défaut 1 (pas de traitement par lots).
parallelism (Facultatif) Nombre maximal de demandes en parallèle. Valeur par défaut 1 (aucun parallélisme).
disableCertificateValidation (Facultatif) Désactiver la validation du certificat SSL. Valeur par défaut "false" (validation activée). Si la valeur est "true", les certificats ne sont pas validés (tous les certificats sont approuvés) et le paramètre "rootCaCertificatePath" est ignoré.
includePubsubMessage (Facultatif) Inclure le message Pub/Sub complet dans la charge utile. Valeur "false" par défaut (seul l'élément de données est inclus dans la charge utile).
tokenKMSEncryptionKey (Facultatif) Clé Cloud KMS permettant de déchiffrer la chaîne du jeton HEC. Si la clé Cloud KMS est fournie, la chaîne du jeton HEC doit être transmise sous forme chiffrée.
rootCaCertificatePath (Facultatif) URL complète du certificat CA racine dans Cloud Storage. Par exemple, gs://mybucket/mycerts/privateCA.crt. Le certificat fourni dans Cloud Storage doit être encodé au format DER et peut être fourni en encodage binaire ou imprimable (base64). Si le certificat est fourni avec un encodage en Base64, il doit être délimité par ------BEGIN CERTIFICATE----- au début et par -----END CERTIFICATE----- à la fin. Si ce paramètre est fourni, ce fichier de certificat CA privé est extrait et ajouté au truststore du nœud de calcul Dataflow pour vérifier le certificat SSL du point de terminaison HEC de Splunk. Si ce paramètre n'est pas fourni, le truststore par défaut est utilisé.

Exécuter le modèle Pub/Sub vers Splunk

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub to Splunk template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Cloud_PubSub_to_Splunk \
    --region REGION_NAME \
    --staging-location TEMP_LOCATION \
    --parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/INPUT_SUBSCRIPTION_NAME,\
token=TOKEN,\
url=URL,\
outputDeadletterTopic=projects/PROJECT_ID/topics/DEADLETTER_TOPIC_NAME,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
batchCount=BATCH_COUNT,\
parallelism=PARALLELISM,\
disableCertificateValidation=DISABLE_VALIDATION,\
rootCaCertificatePath=ROOT_CA_CERTIFICATE_PATH

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • INPUT_SUBSCRIPTION_NAME : nom de l'abonnement Pub/Sub
  • TOKEN : jeton HTTP Event Collector de Splunk
  • URL : chemin d'URL du jeton HTTP Event Collector de Splunk (par exemple, https://splunk-hec-host:8088)
  • DEADLETTER_TOPIC_NAME : nom du sujet Pub/Sub
  • 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.

  • 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).
  • BATCH_COUNT : taille de lot à utiliser pour envoyer plusieurs événements vers Splunk
  • PARALLELISM : nombre de requêtes parallèles à utiliser pour envoyer des événements vers Splunk
  • DISABLE_VALIDATION : true si vous souhaitez désactiver la validation du certificat SSL
  • ROOT_CA_CERTIFICATE_PATH : chemin d'accès au certificat racine de l'autorité de certification dans Cloud Storage (par exemple, gs://your-bucket/privateCA.crt)

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Cloud_PubSub_to_Splunk
{
   "jobName": "JOB_NAME",
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": "gs://your-bucket/temp",
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
   },
   "parameters": {
       "inputSubscription": "projects/PROJECT_ID/subscriptions/INPUT_SUBSCRIPTION_NAME",
       "token": "TOKEN",
       "url": "URL",
       "outputDeadletterTopic": "projects/PROJECT_ID/topics/DEADLETTER_TOPIC_NAME",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "batchCount": "BATCH_COUNT",
       "parallelism": "PARALLELISM",
       "disableCertificateValidation": "DISABLE_VALIDATION",
       "rootCaCertificatePath": "ROOT_CA_CERTIFICATE_PATH"
   }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • INPUT_SUBSCRIPTION_NAME : nom de l'abonnement Pub/Sub
  • TOKEN : jeton HTTP Event Collector de Splunk
  • URL : chemin d'URL du jeton HTTP Event Collector de Splunk (par exemple, https://splunk-hec-host:8088)
  • DEADLETTER_TOPIC_NAME : nom du sujet Pub/Sub
  • 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.

  • 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).
  • BATCH_COUNT : taille de lot à utiliser pour envoyer plusieurs événements vers Splunk
  • PARALLELISM : nombre de requêtes parallèles à utiliser pour envoyer des événements vers Splunk
  • DISABLE_VALIDATION : true si vous souhaitez désactiver la validation du certificat SSL
  • ROOT_CA_CERTIFICATE_PATH : chemin d'accès au certificat racine de l'autorité de certification dans Cloud Storage (par exemple, gs://your-bucket/privateCA.crt)

Pub/Sub vers fichiers Avro dans Cloud Storage

Le modèle Pub/Sub vers fichiers Avro dans Cloud Storage est un pipeline de streaming qui lit les données d'un sujet Pub/Sub et écrit des fichiers Avro dans le bucket Cloud Storage spécifié.

Conditions requises pour ce pipeline :

  • Le sujet Pub/Sub d'entrée doit exister avant l'exécution du pipeline.

Paramètres de modèle

Paramètres Description
inputTopic Sujet Pub/Sub permettant de s'abonner à la consultation de messages. Le nom du sujet doit être au format projects/<project-id>/topics/<topic-name>.
outputDirectory Répertoire de sortie dans lequel les fichiers de sortie Avro seront archivés. Doit inclure / à la fin. Exemple : gs://example-bucket/example-directory/.
avroTempDirectory Répertoire des fichiers Avro temporaires. Doit inclure / à la fin. Par exemple : gs://example-bucket/example-directory/.
outputFilenamePrefix (Facultatif) Préfixe du nom de fichier de sortie pour les fichiers Avro.
outputFilenameSuffix (Facultatif) Suffixe du nom de fichier de sortie pour les fichiers Avro.
outputShardTemplate [Facultatif] Modèle de partition du fichier de sortie. Spécifié en tant que séquences répétées des lettres S ou N. Exemple : SSS-NNN. Celles-ci sont remplacées par le numéro de partition ou par le nombre total de partitions, respectivement. Si ce paramètre n'est pas spécifié, le format du modèle par défaut est W-P-SS-of-NN.

Exécuter le modèle Pub/Sub vers Cloud Storage Avro

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub to Avro Files on Cloud Storage template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Cloud_PubSub_to_Avro \
    --region REGION_NAME \
    --staging-location TEMP_LOCATION \
    --parameters \
inputTopic=projects/PROJECT_ID/topics/TOPIC_NAME,\
outputDirectory=gs://BUCKET_NAME/output/,\
outputFilenamePrefix=FILENAME_PREFIX,\
outputFilenameSuffix=FILENAME_SUFFIX,\
outputShardTemplate=SHARD_TEMPLATE,\
avroTempDirectory=gs://BUCKET_NAME/temp/

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • TOPIC_NAME : nom du sujet Pub/Sub
  • BUCKET_NAME : nom du bucket Cloud Storage
  • FILENAME_PREFIX : préfixe du nom de fichier de sortie préféré
  • FILENAME_SUFFIX : suffixe du nom de fichier de sortie préféré
  • SHARD_TEMPLATE : modèle de partition de sortie préféré

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Cloud_PubSub_to_Avro
{
   "jobName": "JOB_NAME",
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": TEMP_LOCATION,
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
    },
   "parameters": {
       "inputTopic": "projects/PROJECT_ID/topics/TOPIC_NAME",
       "outputDirectory": "gs://BUCKET_NAME/output/",
       "avroTempDirectory": "gs://BUCKET_NAME/temp/",
       "outputFilenamePrefix": "FILENAME_PREFIX",
       "outputFilenameSuffix": "FILENAME_SUFFIX",
       "outputShardTemplate": "SHARD_TEMPLATE"
   }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • TOPIC_NAME : nom du sujet Pub/Sub
  • BUCKET_NAME : nom du bucket Cloud Storage
  • FILENAME_PREFIX : préfixe du nom de fichier de sortie préféré
  • FILENAME_SUFFIX : suffixe du nom de fichier de sortie préféré
  • SHARD_TEMPLATE : modèle de partition de sortie préféré

Pub/Sub vers des fichiers texte dans Cloud Storage

Le modèle Pub/Sub vers texte Cloud Storage est un pipeline de streaming qui lit les enregistrements de Pub/Sub et les enregistre sous forme d'une série de fichiers Cloud Storage au format texte. Le modèle peut être utilisé comme moyen rapide d'enregistrer des données dans Pub/Sub pour une utilisation ultérieure. Par défaut, le modèle génère un nouveau fichier toutes les 5 minutes.

Conditions requises pour ce pipeline :

  • Le sujet Pub/Sub doit exister avant l'exécution.
  • Les messages publiés sur le thème doivent être au format texte.
  • Les messages publiés sur le thème ne doivent contenir aucune nouvelle ligne. Notez que chaque message Pub/Sub est enregistré sur une ligne unique dans le fichier de sortie.

Paramètres de modèle

Paramètres Description
inputTopic Sujet Pub/Sub à partir duquel lire l'entrée. Le nom du sujet doit être au format projects/<project-id>/topics/<topic-name>.
outputDirectory Chemin d'accès et préfixe du nom de fichier pour l'écriture des fichiers de sortie. Par exemple : gs://bucket-name/path/. Cette valeur doit se terminer par une barre oblique.
outputFilenamePrefix Préfixe à placer sur chaque fichier ciblé sur une fenêtre. Par exemple : output-
outputFilenameSuffix Suffixe à placer sur chaque fichier ciblé sur une fenêtre, généralement une extension de fichier telle que .txt ou .csv.
outputShardTemplate Le modèle de segment définit la partie dynamique de chaque fichier ciblé sur une fenêtre. Par défaut, le pipeline utilise un seul segment pour la sortie vers le système de fichiers dans chaque fenêtre. Cela signifie que toutes les données vont atterrir dans un seul fichier par fenêtre. Le outputShardTemplate devient par défaut W-P-SS-of-NNW correspond à la plage de dates de la fenêtre, P correspond aux informations du volet, S correspond au numéro de segment et N au nombre de segments. Dans le cas d'un fichier unique, la partie SS-of-NN de outputShardTemplate sera 00-of-01.

Exécuter le modèle Pub/Sub vers des fichiers texte dans Cloud Storage

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub to Text Files on Cloud Storage template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Cloud_PubSub_to_GCS_Text \
    --region REGION_NAME \
    --staging-location TEMP_LOCATION \
    --parameters \
inputTopic=projects/PROJECT_ID/topics/TOPIC_NAME,\
outputDirectory=gs://BUCKET_NAME/output/,\
outputFilenamePrefix=output-,\
outputFilenameSuffix=.txt

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • TOPIC_NAME : nom de votre sujet Pub/Sub
  • BUCKET_NAME : nom du bucket Cloud Storage

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Cloud_PubSub_to_GCS_Text
{
   "jobName": "JOB_NAME",
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": "TEMP_LOCATION",
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
    },
   "parameters": {
       "inputTopic": "projects/PROJECT_ID/topics/TOPIC_NAME"
       "outputDirectory": "gs://BUCKET_NAME/output/",
       "outputFilenamePrefix": "output-",
       "outputFilenameSuffix": ".txt",
   }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • TOPIC_NAME : nom de votre sujet Pub/Sub
  • BUCKET_NAME : nom du bucket Cloud Storage

Pub/Sub vers MongoDB

Le modèle Pub/Sub vers MongoDB est un pipeline de streaming qui lit les messages encodés au format JSON d'un abonnement Pub/Sub et les écrit dans MongoDB sous forme de documents. Si nécessaire, ce pipeline accepte des transformations supplémentaires qui peuvent être incluses à l'aide d'une fonction JavaScript définie par l'utilisateur. Toute erreur survenue en raison d'une non-concordance du schéma ou d'un format JSON non valide, ou pendant l'exécution de transformations, est enregistrée avec le message d'entrée dans une table BigQuery destinée aux messages non traités. Si la table des enregistrements non traités n'existe pas avant l'exécution, le pipeline la crée automatiquement.

Conditions requises pour ce pipeline :

  • L'abonnement Pub/Sub doit exister et les messages doivent être encodés dans un format JSON valide.
  • Le cluster MongoDB doit exister et être accessible à partir des machines de nœud de calcul Dataflow.

Paramètres de modèle

Paramètres Description
inputSubscription Nom de l'abonnement Pub/Sub. Par exemple : projects/my-project-id/subscriptions/my-subscription-id
mongoDBUri Liste de serveurs MongoDB séparés par une virgule. Par exemple : 192.285.234.12:27017,192.287.123.11:27017
database Base de données dans MongoDB pour stocker la collection. Exemple : my-db.
collection Nom de la collection dans la base de données MongoDB. Exemple : my-collection.
deadletterTable Table BigQuery qui stocke les messages en raison d'échecs (schéma non correspondant, format JSON non valide, etc.). Exemple : project-id:dataset-name.table-name.
javascriptTextTransformGcsPath (Facultatif) URI Cloud Storage du fichier .js contenant la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser. Exemple : gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName (Facultatif) 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.
batchSize (Facultatif) Taille de lot utilisée pour l'insertion par lots de documents dans MongoDB. Valeur par défaut : 1000
batchSizeBytes (Facultatif) Taille du lot en octets. Valeur par défaut : 5242880
maxConnectionIdleTime (Facultatif) Durée maximale d'inactivité autorisée en secondes avant que le délai de connexion ne s'écoule. Valeur par défaut : 60000
sslEnabled (Facultatif) Valeur booléenne indiquant si le protocole SSL est activé pour la connexion à MongoDB. Valeur par défaut : true
ignoreSSLCertificate (Facultatif) Valeur booléenne indiquant si le certificat SSL doit être ignoré. Valeur par défaut : true
withOrdered (Facultatif) Valeur booléenne permettant l'activation d'insertions groupées triées dans MongoDB. Valeur par défaut : true
withSSLInvalidHostNameAllowed (Facultatif) Valeur booléenne indiquant si un nom d'hôte non valide est autorisé pour la connexion SSL. Valeur par défaut : true

Exécuter le modèle Pub/Sub vers MongoDB

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub to MongoDB template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/Cloud_PubSub_to_MongoDB \
    --parameters \
inputSubscription=INPUT_SUBSCRIPTION,\
mongoDBUri=MONGODB_URI,\
database=DATABASE,
collection=COLLECTION,
deadletterTable=UNPROCESSED_TABLE
  

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • JOB_NAME : nom de la tâche de votre choix
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • INPUT_SUBSCRIPTION : abonnement Pub/Sub (par exemple, projects/my-project-id/subscriptions/my-subscription-id)
  • MONGODB_URI : adresses du serveur MongoDB (par exemple, 192.285.234.12:27017,192.287.123.11:27017)
  • DATABASE : nom de la base de données MongoDB (par exemple, users)
  • COLLECTION : nom de la collection MongoDB (par exemple, profiles)
  • UNPROCESSED_TABLE : nom de la table BigQuery (par exemple, your-project:your-dataset.your-table-name)

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 et ses champs d'application d'autorisation, 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": {
          "inputSubscription": "INPUT_SUBSCRIPTION",
          "mongoDBUri": "MONGODB_URI",
          "database": "DATABASE",
          "collection": "COLLECTION",
          "deadletterTable": "UNPROCESSED_TABLE"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Cloud_PubSub_to_MongoDB",
   }
}
  

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • JOB_NAME : nom de la tâche de votre choix
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • INPUT_SUBSCRIPTION : abonnement Pub/Sub (par exemple, projects/my-project-id/subscriptions/my-subscription-id)
  • MONGODB_URI : adresses du serveur MongoDB (par exemple, 192.285.234.12:27017,192.287.123.11:27017)
  • DATABASE : nom de la base de données MongoDB (par exemple, users)
  • COLLECTION : nom de la collection MongoDB (par exemple, profiles)
  • UNPROCESSED_TABLE : nom de la table BigQuery (par exemple, your-project:your-dataset.your-table-name)

Pub/Sub vers Elasticsearch

Le modèle Pub/Sub vers Elasticsearch est un pipeline de streaming qui lit les messages d'un abonnement Pub/Sub, exécute une fonction définie par l'utilisateur et les écrit dans Elasticsearch sous forme de documents. Le modèle Dataflow utilise la fonctionnalité de flux de données d'Elasticsearch pour stocker les données de séries temporelles sur plusieurs index tout en vous attribuant une seule ressource nommée pour les requêtes. Les flux de données sont parfaitement adaptés aux journaux, aux métriques, aux traces et aux autres données générées en continu stockées dans Pub/Sub.

Conditions requises pour ce pipeline

  • L'abonnement Pub/Sub doit exister et les messages doivent être encodés dans un format JSON valide.
  • Hôte Elasticsearch accessible au public sur une instance GCP ou sur Elastic Cloud avec Elasticsearch version 7.0 ou ultérieure. Pour en savoir plus, consultez la section Google Cloud Integration for Elastic (Intégration de Google Cloud pour Elastic).
  • Sujet Pub/Sub pour le résultat de l'erreur.

Paramètres de modèle

Paramètres Description
inputSubscription Abonnement Pub/Sub à consommer. Le nom doit être au format projects/<project-id>/subscriptions/<subscription-name>.
connectionUrl URL Elasticsearch au format https://hostname:[port], ou spécifiez CloudID si vous utilisez Elastic Cloud.
apiKey Clé API encodée en base64 utilisée pour l'authentification.
errorOutputTopic Sujet de sortie Pub/Sub pour la publication d'enregistrements ayant échoué au format projects/<project-id>/topics/<topic-name>.
dataset (Facultatif) Type de journaux envoyés via Pub/Sub pour lesquels nous disposons d'un tableau de bord prêt à l'emploi. Les valeurs de types de journaux connues sont audit, vpcflow et firewall. Valeur par défaut : pubsub
namespace (Facultatif) Regroupement arbitraire, tel qu'un environnement (développement, production ou QA), une équipe ou une unité commerciale stratégique. Valeur par défaut : default
batchSize (Facultatif) Taille de lot en nombre de documents. Valeur par défaut : 1000
batchSizeBytes (Facultatif) Taille de lot en octets. Valeur par défaut : 5242880 (5 Mo).
maxRetryAttempts (Facultatif) Nombre maximal de nouvelles tentatives, doit être supérieur 0. Valeur par défaut : no retries
maxRetryDuration (Facultatif) Durée maximale de la nouvelle tentative en millisecondes, doit être supérieure à 0. Valeur par défaut : no retries
javascriptTextTransformGcsPath (Facultatif) URI Cloud Storage du fichier .js contenant la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser. Exemple : gs://my-bucket/my-udfs/my_file.js.
javascriptTextTransformFunctionName (Facultatif) 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.

Exécuter le modèle Pub/Sub vers Elasticsearch

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Pub/Sub to Elasticsearch template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/PubSub_to_Elasticsearch \
    --parameters \
inputSubscription=SUBSCRIPTION_NAME,\
connectionUrl=CONNECTION_URL,\
dataset=DATASET,\
namespace=NAMESPACE,\
apiKey=APIKEY,\
errorOutputTopic=ERROR_OUTPUT_TOPIC
  

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • ERROR_OUTPUT_TOPIC : sujet Pub/Sub pour le résultat d'erreur
  • SUBSCRIPTION_NAME : nom de votre abonnement Pub/Sub
  • CONNECTION_URL : URL Elasticsearch
  • DATASET : type de journal
  • NAMESPACE : espace de noms pour un ensemble de données
  • APIKEY : clé API encodée en base64 pour l'authentification

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 et ses champs d'application d'autorisation, 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": {
          "inputSubscription": "SUBSCRIPTION_NAME",
          "connectionUrl": "CONNECTION_URL",
          "dataset": "DATASET",
          "namespace": "NAMESPACE",
          "apiKey": "APIKEY",
          "errorOutputTopic": "ERROR_OUTPUT_TOPIC"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/PubSub_to_Elasticsearch",
   }
}
  

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • ERROR_OUTPUT_TOPIC : sujet Pub/Sub pour le résultat d'erreur
  • SUBSCRIPTION_NAME : nom de votre abonnement Pub/Sub
  • CONNECTION_URL : URL Elasticsearch
  • DATASET : type de journal
  • NAMESPACE : espace de noms pour un ensemble de données
  • APIKEY : clé API encodée en base64 pour l'authentification

Datastream vers Cloud Spanner

Le modèle Datastream vers Cloud Spanner est un pipeline de streaming qui lit les événements Datastream d'un bucket Cloud Storage et les écrit dans une base de données Cloud Spanner. Il est destiné à la migration de données de sources Datastream vers Cloud Spanner.

Toutes les tables requises pour la migration doivent exister dans la base de données Cloud Spanner de destination avant l'exécution du modèle. Par conséquent, la migration du schéma d'une base de données source vers Cloud Spanner doit être terminée avant la migration des données. Des données peuvent exister dans les tables avant la migration. Ce modèle ne propage pas les modifications du schéma Datastream dans la base de données Cloud Spanner.

La cohérence des données n'est garantie à la fin de la migration que lorsque toutes les données ont été écrites dans Cloud Spanner. Pour stocker des informations de tri pour chaque enregistrement écrit dans Cloud Spanner, ce modèle crée une table supplémentaire (appelée "shadow table" ou table fictive) pour chaque table de la base de données Cloud Spanner. Cela permet de garantir la cohérence à la fin de la migration. Les tables fictives ne sont pas supprimées après la migration et peuvent être utilisées à des fins de validation à la fin de la migration.

Toutes les erreurs qui se produisent pendant l'opération, telles que les incohérences de schéma, les fichiers JSON non valides ou les erreurs résultant de l'exécution des transformations, sont enregistrées dans une file d'attente d'erreurs. La file d'attente d'erreurs est un dossier Cloud Storage qui stocke tous les événements Datastream ayant rencontré des erreurs ainsi que le motif de l'erreur au format texte. Les erreurs peuvent être temporaires ou permanentes, et sont stockées dans des dossiers Cloud Storage appropriés dans la file d'attente d'erreurs. Les erreurs temporaires font l'objet de nouvelles tentatives automatiques, contrairement aux erreurs permanentes. En cas d'erreurs permanentes, vous avez la possibilité de corriger les événements de modification et de les déplacer vers le bucket pouvant faire l'objet de nouvelles tentatives pendant l'exécution du modèle.

Conditions requises pour ce pipeline :

  • Un flux Datastream dans l'état En cours d'exécution ou Non démarré.
  • Un bucket Cloud Storage dans lequel les événements Datastream sont répliqués
  • Une base de données Cloud Spanner avec tables existantes. Ces tables peuvent être vides ou contenir des données.

Paramètres de modèle

Paramètres Description
inputFilePattern Emplacement des fichiers Datastream à répliquer dans Cloud Storage. Il s'agit généralement du chemin racine d'un flux.
streamName Nom ou modèle du flux à interroger pour obtenir des informations de schéma et le type de source.
instanceId Instance Cloud Spanner sur laquelle les modifications sont répliquées.
databaseId Base de données Cloud Spanner dans laquelle les modifications sont répliquées.
projectId ID du projet Cloud Spanner.
deadLetterQueueDirectory (Facultatif) Il s'agit du chemin d'accès au fichier permettant de stocker la sortie de la file d'attente d'erreurs. La valeur par défaut est un répertoire situé sous l'emplacement temporaire de la tâche Dataflow.
inputFileFormat (Facultatif) Format du fichier de sortie généré par Datastream. Par exemple, avro,json. Par défaut, avro.
shadowTablePrefix (Facultatif) Préfixe utilisé pour nommer les tables fantômes. Valeur par défaut : shadow_

Exécuter le modèle Datastream vers Cloud Spanner

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Cloud Datastream to Spanner template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud beta dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/Datastream_to_CloudSpanner \
    --parameters \
inputFilePattern=GCS_FILE_PATH,\
streamName=STREAM_NAME,\
instanceId=CLOUDSPANNER_INSTANCE,\
databaseId=CLOUDSPANNER_DATABASE,\
deadLetterQueueDirectory=DLQ
  

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • GCS_FILE_PATH : chemin d'accès Cloud Storage utilisé pour stocker les événements de flux de données. Par exemple : gs://bucket/path/to/data/
  • CLOUDSPANNER_INSTANCE : votre instance Cloud Spanner
  • CLOUDSPANNER_DATABASE : votre base de données Cloud Spanner.
  • DLQ : chemin d'accès Cloud Storage au répertoire de la file d'attente d'erreurs.

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 et ses champs d'application d'autorisation, 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": {

inputFilePattern=GCS_FILE_PATH,\
streamName=STREAM_NAME,\
instanceId=CLOUDSPANNER_INSTANCE,\
databaseId=CLOUDSPANNER_DATABASE,\
deadLetterQueueDirectory=DLQ
          "inputFilePattern": "GCS_FILE_PATH",
          "streamName": "STREAM_NAME"
          "instanceId": "CLOUDSPANNER_INSTANCE"
          "databaseId": "CLOUDSPANNER_DATABASE"
          "deadLetterQueueDirectory": "DLQ"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Datastream_to_CloudSpanner",
   }
}
  

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • GCS_FILE_PATH : chemin d'accès Cloud Storage utilisé pour stocker les événements de flux de données. Par exemple : gs://bucket/path/to/data/
  • CLOUDSPANNER_INSTANCE : votre instance Cloud Spanner
  • CLOUDSPANNER_DATABASE : votre base de données Cloud Spanner.
  • DLQ : chemin d'accès Cloud Storage au répertoire de la file d'attente d'erreurs.

Fichiers texte dans Cloud Storage vers BigQuery (Flux)

Les fichiers texte dans Cloud Storage vers BigQuery sont un pipeline de streaming qui vous permet de lire des fichiers texte stockés dans Cloud Storage, de les transformer à l'aide de la fonction JavaScript définie par l'utilisateur (UDF) que vous fournissez, et d'ajouter le résultat à BigQuery.

Le pipeline fonctionne indéfiniment et doit être arrêté manuellement via une annulation et non un drainage, en raison de son utilisation de la transformation Watch qui est une fonction DoFn non compatible avec le drainage.

Conditions requises pour ce pipeline :

  • Créez un fichier JSON décrivant le schéma de la table de sortie dans BigQuery.

    Assurez-vous qu'il existe un tableau JSON de niveau supérieur intitulé BigQuery Schema et que son contenu suit le modèle {"name": "COLUMN_NAME", "type": "DATA_TYPE"}. Exemple :

    {
      "BigQuery Schema": [
        {
          "name": "location",
          "type": "STRING"
        },
        {
          "name": "name",
          "type": "STRING"
        },
        {
          "name": "age",
          "type": "STRING"
        },
        {
          "name": "color",
          "type": "STRING",
          "mode": "REQUIRED"
        },
        {
          "name": "coffee",
          "type": "STRING",
          "mode": "REQUIRED"
        }
      ]
    }
    
  • Créez un fichier JavaScript (.js) à l'aide de la fonction définie par l'utilisateur (UDF) qui fournit la logique pour transformer les lignes de texte. Notez que votre fonction doit renvoyer une chaîne JSON.

    Par exemple, cette fonction divise chaque ligne d'un fichier CSV et renvoie une chaîne JSON après avoir transformé les valeurs.

    function transform(line) {
    var values = line.split(',');
    
    var obj = new Object();
    obj.location = values[0];
    obj.name = values[1];
    obj.age = values[2];
    obj.color = values[3];
    obj.coffee = values[4];
    var jsonString = JSON.stringify(obj);
    
    return jsonString;
    }
    

Paramètres de modèle

Paramètres Description
javascriptTextTransformGcsPath URI Cloud Storage du fichier .js contenant la fonction JavaScript définie par l'utilisateur que vous souhaitez utiliser. Exemple : gs://my-bucket/my-udfs/my_file.js.
JSONPath Emplacement Cloud Storage de votre fichier de schéma BigQuery, décrit comme fichier JSON. Exemple : gs://path/to/my/schema.json.
javascriptTextTransformFunctionName 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.
outputTable Nom complet de la table BigQuery. Par exemple : my-project:dataset.table
inputFilePattern Emplacement Cloud Storage du texte que vous souhaitez traiter. Exemple : gs://my-bucket/my-files/text.txt.
bigQueryLoadingTemporaryDirectory Répertoire temporaire pour le processus de chargement de BigQuery. Par exemple : gs://my-bucket/my-files/temp_dir
outputDeadletterTable Table des messages qui n'ont pas pu atteindre la table de sortie. Exemple : my-project:dataset.my-unprocessed-table. Si elle n'existe pas, elle sera créée lors de l'exécution du pipeline. Si ce paramètre n'est pas spécifié, <outputTableSpec>_error_records est utilisé à la place.

Exécuter le modèle Texte Cloud Storage vers BigQuery (Flux)

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Text Files on Cloud Storage to BigQuery template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Stream_GCS_Text_to_BigQuery \
    --region REGION_NAME \
    --staging-location TEMP_LOCATION \
    --parameters \
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
JSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
inputFilePattern=PATH_TO_TEXT_DATA,\
outputTable=BIGQUERY_TABLE,\
outputDeadletterTable=BIGQUERY_UNPROCESSED_TABLE,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.

  • PATH_TO_BIGQUERY_SCHEMA_JSON : chemin d'accès Cloud Storage au fichier JSON contenant la définition du schéma
  • 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).
  • PATH_TO_TEXT_DATA : chemin d'accès Cloud Storage à votre ensemble de données texte
  • BIGQUERY_TABLE : nom de votre table BigQuery.
  • BIGQUERY_UNPROCESSED_TABLE : nom de votre table BigQuery pour les messages non traités
  • PATH_TO_TEMP_DIR_ON_GCS : chemin d'accès Cloud Storage au répertoire temporaire

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Stream_GCS_Text_to_BigQuery
{
   "jobName": "JOB_NAME",
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": "TEMP_LOCATION",
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
    },
   "parameters": {
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "JSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "inputFilePattern":"PATH_TO_TEXT_DATA",
       "outputTable":"BIGQUERY_TABLE",
       "outputDeadletterTable":"BIGQUERY_UNPROCESSED_TABLE",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS"
   }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • 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 est myTransform. Pour obtenir des exemples de fonctions JavaScript définies par l'utilisateur, consultez la page Exemples de fonctions définies par l'utilisateur.

  • PATH_TO_BIGQUERY_SCHEMA_JSON : chemin d'accès Cloud Storage au fichier JSON contenant la définition du schéma
  • 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).
  • PATH_TO_TEXT_DATA : chemin d'accès Cloud Storage à votre ensemble de données texte
  • BIGQUERY_TABLE : nom de votre table BigQuery.
  • BIGQUERY_UNPROCESSED_TABLE : nom de votre table BigQuery pour les messages non traités
  • PATH_TO_TEMP_DIR_ON_GCS : chemin d'accès Cloud Storage au répertoire temporaire

Fichiers texte sur Cloud Storage vers Pub/Sub (Flux)

Le modèle crée un pipeline de streaming qui interroge en permanence les nouveaux fichiers texte chargés dans Cloud Storage, lit chaque fichier ligne par ligne et publie des chaînes dans un sujet Pub/Sub. Le modèle publie les enregistrements dans un fichier délimité par une nouvelle ligne contenant des enregistrements JSON ou un fichier CSV dans un sujet Pub/Sub pour un traitement en temps réel. Vous pouvez utiliser ce modèle pour relire les données dans Pub/Sub.

Le pipeline s'exécute indéfiniment et doit être arrêté manuellement via une annulation et non un drainage, en raison de son utilisation de la transformation "Watch", qui est une fonction "SplittableDoFn" qui n'est pas compatible avec le drainage.

Actuellement, l'intervalle d'interrogation est fixé à 10 secondes. Ce modèle ne définit aucun horodatage sur les enregistrements individuels. Par conséquent, l'heure de l'événement sera égale à l'heure de publication pendant l'exécution. Si votre pipeline dépend d'une heure d'événement précise pour le traitement, ne l'utilisez pas.

Conditions requises pour ce pipeline :

  • Les fichiers d'entrée doivent être au format CSV ou JSON délimité par une nouvelle ligne. Les enregistrements couvrant plusieurs lignes dans les fichiers sources peuvent entraîner des problèmes en aval, car chaque ligne dans les fichiers sera publiée sous forme de message à Pub/Sub.
  • Le sujet Pub/Sub doit exister avant l'exécution.
  • Le pipeline fonctionne indéfiniment et doit être terminé manuellement.

Paramètres de modèle

Paramètres Description
inputFilePattern Modèle de fichier d'entrée à lire. Par exemple, gs://bucket-name/files/*.json ou gs://bucket-name/path/*.csv.
outputTopic Sujet d'entrée Pub/Sub dans lequel écrire. Le nom doit être au format projects/<project-id>/topics/<topic-name>.

Exécuter le modèle de fichiers texte dans Cloud Storage vers Pub/Sub (Flux)

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Text Files on Cloud Storage to Pub/Sub (Stream) template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Stream_GCS_Text_to_Cloud_PubSub \
    --region REGION_NAME\
    --staging-location TEMP_LOCATION\
    --parameters \
inputFilePattern=gs://BUCKET_NAME/FILE_PATTERN,\
outputTopic=projects/PROJECT_ID/topics/TOPIC_NAME

Remplacez les éléments suivants :

  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • TOPIC_NAME : nom de votre sujet Pub/Sub
  • BUCKET_NAME : nom de votre bucket Cloud Storage
  • FILE_PATTERN : modèle de fichier glob à lire dans le bucket Cloud Storage (par exemple, path/*.csv)

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Stream_GCS_Text_to_Cloud_PubSub
{
   "jobName": "JOB_NAME",
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": "gs://your-bucket/temp",
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
    },
   "parameters": {
       "inputFilePattern": "gs://BUCKET_NAME/FILE_PATTERN",
       "outputTopic": "projects/PROJECT_ID/topics/TOPIC_NAME"
   }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • TOPIC_NAME : nom de votre sujet Pub/Sub
  • BUCKET_NAME : nom de votre bucket Cloud Storage
  • FILE_PATTERN : modèle de fichier glob à lire dans le bucket Cloud Storage (par exemple, path/*.csv)

Masquage/Tokenisation de données de Cloud Storage vers BigQuery (à l'aide de Cloud DLP)

Le modèle Masquage/Tokenisation de données de Cloud Storage vers BigQuery (à l'aide de Cloud DLP) est un pipeline de streaming qui lit les fichiers CSV d'un bucket Cloud Storage, appelle l'API Cloud Data Loss Prevention (Cloud DLP) pour les anonymiser puis écrit les données anonymisées dans la table BigQuery spécifiée. Ce modèle accepte l'utilisation d'un modèle d'inspection Cloud DLP et d'un modèle de suppression d'identification Cloud DLP. Cela permet aux utilisateurs d'inspecter du texte pour identifier les données potentiellement sensibles, et de supprimer l'identification des données sensibles détectées. Ils ont également la possibilité de supprimer l'identification de données structurées lorsque des colonnes sont spécifiées et qu'aucune inspection n'est nécessaire.

Conditions requises pour ce pipeline :

  • Les données d'entrée à tokeniser doivent exister.
  • Les modèles Cloud DLP doivent exister (par exemple, DeidentifyTemplate et InspectTemplate). Pour en savoir plus, consultez la page Modèles Cloud DLP.
  • L'ensemble de données BigQuery doit exister.

Paramètres de modèle

Paramètres Description
inputFilePattern Fichiers csv à partir desquels lire les enregistrements de données d'entrée. Les caractères génériques sont également acceptés. Par exemple, gs://mybucket/my_csv_filename.csv ou gs://mybucket/file-*.csv.
dlpProjectId ID du projet Cloud DLP propriétaire de la ressource API Cloud DLP. Il peut s'agir du projet propriétaire des modèles Cloud DLP ou d'un autre projet. Par exemple, my_dlp_api_project.
deidentifyTemplateName Modèle d'anonymisation Cloud DLP à utiliser pour les requêtes d'API, spécifié à l'aide du schéma projects/{template_project_id}/deidentifyTemplates/{deIdTemplateId}. Exemple :projects/my_project/deidentifyTemplates/100
datasetName Ensemble de données BigQuery pour l'envoi des résultats tokenisés.
batchSize Taille du lot pour l'envoi des données à inspecter et/ou à détokeniser. Dans le cas d'un fichier csv, batchSize correspond au nombre de lignes dans un lot. Les utilisateurs doivent déterminer la taille du lot en fonction de la taille des enregistrements et de la taille du fichier. Notez que l'API Cloud DLP limite la taille de la charge utile à 524 Ko par appel d'API.
inspectTemplateName (Facultatif) Modèle d'inspection Cloud DLP à utiliser pour les requêtes d'API, spécifié à l'aide du schéma projects/{template_project_id}/identifyTemplates/{idTemplateId}. Exemple :projects/my_project/identifyTemplates/100

Exécuter le modèle Masquage/Tokenisation de données de Cloud Storage vers BigQuery (à l'aide de Cloud DLP)

Console

  1. Accédez à la page Dataflow Créer une tâche à partir d'un modèle.
  2. Accéder à la page Créer une tâche à partir d'un modèle
  3. Dans le champ Nom de la tâche, saisissez un nom de tâche unique.
  4. Facultatif : pour Point de terminaison régional, sélectionnez une valeur dans le menu déroulant. Le point de terminaison régional par défaut est us-central1.

    Pour obtenir la liste des régions dans lesquelles vous pouvez exécuter une tâche Dataflow, consultez la page Emplacements Dataflow.

  5. Dans le menu déroulant Modèle Dataflow, sélectionnez the Data Masking/Tokenization from Cloud Storage to BigQuery (using Cloud DLP) template.
  6. Dans les champs fournis, saisissez vos valeurs de paramètres.
  7. Cliquez sur Run Job (Exécuter la tâche).

gcloud

Dans l'interface système ou le terminal, exécutez le modèle :

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/Stream_DLP_GCS_Text_to_BigQuery \
    --region REGION_NAME \
    --staging-location TEMP_LOCATION \
    --parameters \
inputFilePattern=INPUT_DATA,\
datasetName=DATASET_NAME,\
batchSize=BATCH_SIZE_VALUE,\
dlpProjectId=DLP_API_PROJECT_ID,\
deidentifyTemplateName=projects/TEMPLATE_PROJECT_ID/deidentifyTemplates/DEIDENTIFY_TEMPLATE,\
inspectTemplateName=projects/TEMPLATE_PROJECT_ID/identifyTemplates/INSPECT_TEMPLATE_NUMBER

Remplacez les éléments suivants :

  • DLP_API_PROJECT_ID : ID de votre projet dans l'API Cloud DLP
  • JOB_NAME : nom de la tâche de votre choix
  • REGION_NAME : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • INPUT_DATA : chemin d'accès de votre fichier d'entrée
  • DEIDENTIFY_TEMPLATE : numéro de modèle d'anonymisation Cloud DLP
  • DATASET_NAME : nom de l'ensemble de données BigQuery
  • INSPECT_TEMPLATE_NUMBER : numéro du modèle d'inspection Cloud DLP
  • BATCH_SIZE_VALUE : taille du lot (nombre de lignes par API pour les données CSV)

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 et ses champs d'application d'autorisation, consultez la section projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates/VERSION/Stream_DLP_GCS_Text_to_BigQuery
{
   "jobName": "JOB_NAME",
   "environment": {
       "bypassTempDirValidation": false,
       "tempLocation": "TEMP_LOCATION",
       "ipConfiguration": "WORKER_IP_UNSPECIFIED",
       "additionalExperiments": []
   },
   "parameters": {
      "inputFilePattern":INPUT_DATA,
      "datasetName": "DATASET_NAME",
      "batchSize": "BATCH_SIZE_VALUE",
      "dlpProjectId": "DLP_API_PROJECT_ID",
      "deidentifyTemplateName": "projects/TEMPLATE_PROJECT_ID/deidentifyTemplates/DEIDENTIFY_TEMPLATE",
      "inspectTemplateName": "projects/TEMPLATE_PROJECT_ID/identifyTemplates/INSPECT_TEMPLATE_NUMBER"
   }
}

Remplacez les éléments suivants :

  • PROJECT_ID : ID du projet Cloud dans lequel vous souhaitez exécuter la tâche Dataflow
  • DLP_API_PROJECT_ID : ID de votre projet dans l'API Cloud DLP
  • JOB_NAME : nom de la tâche de votre choix
  • LOCATION : point de terminaison régional où vous souhaitez déployer votre tâche Dataflow, par exemple us-central1
  • VERSION : version du modèle que vous souhaitez utiliser

    Vous 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/latest/
    • Le nom de la version, par exemple :2021-09-20-00_RC00, pour utiliser une version spécifique du modèle, qui peut être imbriquée dans le dossier parent daté du bucket :gs://dataflow-templates/
  • TEMP_LOCATION : emplacement de l'écriture de fichiers temporaires (par exemple, gs://your-bucket/temp)
  • INPUT_DATA : chemin d'accès de votre fichier d'entrée
  • DEIDENTIFY_TEMPLATE : numéro de modèle d'anonymisation Cloud DLP
  • DATASET_NAME : nom de l'ensemble de données BigQuery
  • INSPECT_TEMPLATE_NUMBER : numéro du modèle d'inspection Cloud DLP
  • BATCH_SIZE_VALUE : taille du lot (nombre de lignes par API pour les données CSV)