Créer un pipeline Dataflow à l'aide de Go

Créer un pipeline Dataflow à l'aide de Go

Cette page vous explique comment utiliser le SDK Apache Beam pour Go afin de créer un programme qui définit un pipeline. Ensuite, vous exécutez le pipeline en local et sur le service Dataflow. Pour une présentation du pipeline WordCount, regardez la vidéo How to use WordCount in Apache Beam (Comment utiliser WordCount dans Apache Beam).

Avant de commencer

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
2. Install the Google Cloud CLI.

3. To initialize the gcloud CLI, run the following command:

   gcloud init

4. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. Make sure that billing is enabled for your Google Cloud project.

6. Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager APIs:

   gcloud services enable dataflow compute_component logging storage_component storage_api cloudresourcemanager.googleapis.com

7. Create local authentication credentials for your user account:

   gcloud auth application-default login

8. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser

   gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

   • Replace PROJECT_ID with your project ID.

   • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

   • Replace ROLE with each individual role.
9. Install the Google Cloud CLI.

10. To initialize the gcloud CLI, run the following command:

    gcloud init

11. Create or select a Google Cloud project.

    Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

    • Create a Google Cloud project:

      gcloud projects create PROJECT_ID

      Replace PROJECT_ID with a name for the Google Cloud project you are creating.

    • Select the Google Cloud project that you created:

      gcloud config set project PROJECT_ID

      Replace PROJECT_ID with your Google Cloud project name.

12. Make sure that billing is enabled for your Google Cloud project.

13. Enable the Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager APIs:

    gcloud services enable dataflow compute_component logging storage_component storage_api cloudresourcemanager.googleapis.com

14. Create local authentication credentials for your user account:

    gcloud auth application-default login

15. Grant roles to your user account. Run the following command once for each of the following IAM roles: roles/iam.serviceAccountUser

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

    • Replace PROJECT_ID with your project ID.

    • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

    • Replace ROLE with each individual role.

16. Attribuez des rôles à votre compte de service Compute Engine par défaut. Exécutez la commande suivante une fois pour chacun des rôles IAM suivants :

    • roles/dataflow.admin
    • roles/dataflow.worker
    • roles/storage.objectAdmin

    gcloud projects add-iam-policy-binding PROJECT_ID --member="serviceAccount:[e-mail]" --role=SERVICE_ACCOUNT_ROLE

    • Remplacez PROJECT_ID par l'ID du projet.
    • Remplacez PROJECT_NUMBER par votre numéro de projet. Pour trouver votre numéro de projet, consultez la section Identifier des projets ou utilisez la commande gcloud projects describe.
    • Remplacez SERVICE_ACCOUNT_ROLE par chaque rôle individuel.
17. Create a Cloud Storage bucket and configure it as follows:
    • Set the storage class to S (Standard).
    • Définissez l'emplacement de stockage sur : US (États-Unis).
    • Remplacez BUCKET_NAME par un nom de bucket unique. N'incluez aucune information sensible dans le nom des buckets, car leur espace de noms est global et visible par tous.

    gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US

    • Copiez l'ID du projet Google Cloud et le nom du bucket Cloud Storage. Vous aurez besoin de ces valeurs plus loin dans ce guide de démarrage rapide.

Configurer l'environnement de développement

Le SDK Apache Beam est un modèle de programmation Open Source pour les pipelines de données. Vous définissez un pipeline avec un programme Apache Beam, puis l'exécutez à l'aide d'un exécuteur tel que Dataflow.

Nous vous recommandons d'utiliser la dernière version de Go lorsque vous utilisez le SDK Apache Beam pour Go. Si la dernière version de Go n'est pas installée, utilisez le guide de téléchargement et d'installation de Go pour télécharger et installer Go sur votre système d'exploitation spécifique.

Pour vérifier la version de Go que vous avez installée, exécutez la commande suivante dans votre terminal local :

go version

Exécuter l'exemple de décompte de mots Beam

Le SDK Apache Beam pour Go inclut un exemple de pipeline wordcount. L'exemple wordcount effectue les opérations suivantes :

1. Lit un fichier texte en entrée. Par défaut, il lit un fichier texte situé dans un bucket Cloud Storage avec le nom de ressource gs://dataflow-samples/shakespeare/kinglear.txt.
2. Analyser chaque ligne en mots
3. Calculer la fréquence des mots en fonction des mots tokenisé.

Pour exécuter la dernière version de l'exemple Beam wordcount sur votre ordinateur local, utilisez la commande suivante. L'option input spécifie le fichier à lire, et l'option output spécifie le nom de fichier à utiliser pour délivrer le décompte de fréquence en sortie.

go run github.com/apache/beam/sdks/v2/go/examples/wordcount@latest \
  --input gs://dataflow-samples/shakespeare/kinglear.txt \
  --output outputs

Une fois l'exécution du pipeline terminée, affichez les résultats :

more outputs*

Pour quitter, appuyez sur q.

Modifier le code du pipeline

Le pipeline wordcount de Beam fait la distinction entre les mots en majuscules et en minuscules. Les étapes suivantes montrent comment créer votre propre module Go, modifier le pipeline wordcount afin qu'il ne soit pas sensible à la casse, et l'exécuter sur Dataflow.

Créer un module Go

Pour modifier le code du pipeline, procédez comme suit.

1. Créez un répertoire pour votre module Go à l'emplacement de votre choix :

   mkdir wordcount
   cd wordcount

2. Créez un module Go. Pour cet exemple, utilisez example/dataflow comme chemin de module.

   go mod init example/dataflow

3. Téléchargez la dernière copie du code wordcount à partir du dépôt GitHub d'Apache Beam. Placez ce fichier dans le répertoire wordcount que vous avez créé.

4. Si vous utilisez un système d'exploitation non-Linux, vous devez obtenir le package Go unix. Ce package est nécessaire pour exécuter des pipelines sur le service Dataflow.

   go get -u golang.org/x/sys/unix

5. Assurez-vous que le fichier go.mod correspond au code source du module :

   go mod tidy

Exécuter le pipeline non modifié

Vérifiez que le pipeline wordcount non modifié s'exécute localement.

1. Depuis le terminal, créez et exécutez le pipeline localement :

    go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
        --output outputs

2. Affichez les résultats :

    more outputs*

3. Pour quitter, appuyez sur q.

Modifier le code du pipeline

Pour modifier le pipeline afin qu'il ne soit pas sensible à la casse, modifiez le code pour appliquer la fonction strings.ToLower à tous les mots.

1. Dans l'éditeur de votre choix, ouvrez le fichier wordcount.go.

2. Examinez le bloc init (les commentaires ont été supprimés pour plus de clarté):

    func init() {
      register.DoFn3x0[context.Context, string, func(string)](&extractFn{})
      register.Function2x1(formatFn)
      register.Emitter1[string]()
    }
   

3. Ajoutez une ligne pour enregistrer la fonction strings.ToLower :

    func init() {
      register.DoFn3x0[context.Context, string, func(string)](&extractFn{})
      register.Function2x1(formatFn)
      register.Emitter1[string]()
      register.Function1x1(strings.ToLower)
    }
   

4. Examinez la fonction CountWords :

    func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection {
      s = s.Scope("CountWords")
   
      // Convert lines of text into individual words.
      col := beam.ParDo(s, &extractFn{SmallWordLength: *smallWordLength}, lines)
   
      // Count the number of times each word occurs.
      return stats.Count(s, col)
    }
   

5. Pour convertir les mots en minuscules, ajoutez une opération ParDo qui applique strings.ToLower à chaque mot :

    func CountWords(s beam.Scope, lines beam.PCollection) beam.PCollection {
      s = s.Scope("CountWords")
   
      // Convert lines of text into individual words.
      col := beam.ParDo(s, &extractFn{SmallWordLength: *smallWordLength}, lines)
   
      // Map all letters to lowercase.
      lowercaseWords := beam.ParDo(s, strings.ToLower, col)
   
      // Count the number of times each word occurs.
      return stats.Count(s, lowercaseWords)
    }
   

6. Enregistrez le fichier.

Exécuter le pipeline mis à jour en local

Exécutez votre pipeline wordcount mis à jour localement et vérifiez que le résultat a changé.

1. Créez et exécutez le pipeline wordcount modifié :

    go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
        --output outputs

2. Affichez les résultats du pipeline modifié. Tous les mots doivent être en minuscules.

    more outputs*

3. Pour quitter, appuyez sur q.

Exécuter le pipeline sur le service Dataflow

Pour exécuter l'exemple wordcount mis à jour sur le service Dataflow, utilisez la commande suivante :

go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner dataflow \
    --project PROJECT_ID \
    --region DATAFLOW_REGION \
    --staging_location gs://BUCKET_NAME/binaries/

Remplacez les éléments suivants :

• BUCKET_NAME : nom du bucket Cloud Storage.

• PROJECT_ID : ID de projet Google Cloud.

• DATAFLOW_REGION : région dans laquelle vous souhaitez déployer la tâche Dataflow. Exemple :europe-west1 Pour obtenir la liste des emplacements disponibles, voir Emplacements Dataflow. L'option --region remplace la région par défaut définie dans le serveur de métadonnées, sur votre client local ou dans les variables d'environnement.

Remarque : Pour spécifier un compte de service de nœud de calcul géré par l'utilisateur, incluez l'option de pipeline --service_account_email. Les comptes de service de nœud de calcul gérés par l'utilisateur sont recommandés pour les charges de travail de production. Si vous ne spécifiez pas de compte de service de nœud de calcul lorsque vous créez un job, Dataflow utilise le compte de service Compute Engine par défaut.

Afficher les résultats

Vous pouvez afficher la liste de vos tâches Dataflow dans la console Google Cloud. Dans la console Google Cloud, accédez à la page Tâches de Dataflow.

Accéder aux tâches

La page Tâches affiche les détails de votre tâche wordcount, y compris l'état En cours d'exécution, puis Réussite.

Lorsque vous exécutez un pipeline à l'aide de Dataflow, vos résultats sont stockés dans un bucket Cloud Storage. Affichez les résultats de sortie en utilisant la console Google Cloud ou votre terminal local.

Console

Pour afficher vos résultats dans la console Google Cloud, accédez à la page Buckets de Cloud Storage.

Accéder à la page "Buckets"

Dans la liste des buckets de votre projet, cliquez sur le bucket de stockage que vous avez créé précédemment. Les fichiers de sortie créés par votre tâche s'affichent dans le répertoire results.

Terminal

Affichez les résultats à partir de votre terminal ou à l'aide de Cloud Shell.

1. Pour répertorier les fichiers de sortie, utilisez la commande gcloud storage ls :

   gcloud storage ls gs://BUCKET_NAME/results/outputs* --long

   Remplacez BUCKET_NAME par le nom du bucket Cloud Storage de sortie spécifié.

2. Pour afficher les résultats dans les fichiers de sortie, utilisez la commande gcloud storage cat :

   gcloud storage cat gs://BUCKET_NAME/results/outputs*

Effectuer un nettoyage

Pour éviter que les ressources utilisées sur cette page ne soient facturées sur votre compte Google Cloud, supprimez le projet Google Cloud contenant les ressources.

Remarque : Si vous avez suivi ce guide de démarrage rapide dans un nouveau projet, vous pouvez supprimer le projet.

1. In the Google Cloud console, go to the Cloud Storage Buckets page.

   Go to Buckets

2. Click the checkbox for the bucket that you want to delete.
3. To delete the bucket, click Delete, and then follow the instructions.

4. Si vous conservez votre projet, révoquez les rôles que vous avez accordés au compte de service Compute Engine par défaut. Exécutez la commande suivante une fois pour chacun des rôles IAM suivants :

   • roles/dataflow.admin
   • roles/dataflow.worker
   • roles/storage.objectAdmin

   gcloud projects remove-iam-policy-binding PROJECT_ID \
       --member=serviceAccount:[e-mail] \
       --role=SERVICE_ACCOUNT_ROLE

5. Optional: Revoke the authentication credentials that you created, and delete the local credential file.

   gcloud auth application-default revoke

6. Optional: Revoke credentials from the gcloud CLI.

   gcloud auth revoke

Étapes suivantes

• Modèle de programmation pour Apache Beam
• Définir les options de pipeline
• Documentation de référence sur les options de pipeline
• Déployer un pipeline