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
- Connectez-vous à votre compte Google Cloud. Si vous débutez sur Google Cloud, créez un compte pour évaluer les performances de nos produits en conditions réelles. Les nouveaux clients bénéficient également de 300 $ de crédits gratuits pour exécuter, tester et déployer des charges de travail.
- Install the Google Cloud CLI.
-
To initialize the gcloud CLI, run the following command:
gcloud init
-
Create or select a Google Cloud 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.
-
-
Make sure that billing is enabled for your Google Cloud project.
-
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 -
Create local authentication credentials for your user account:
gcloud auth application-default login
-
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_IDENTIFIER" --role=ROLE
- Replace
PROJECT_ID
with your project ID. -
Replace
USER_IDENTIFIER
with the identifier for your user account. For example,user:myemail@example.com
. - Replace
ROLE
with each individual role.
- Replace
- Install the Google Cloud CLI.
-
To initialize the gcloud CLI, run the following command:
gcloud init
-
Create or select a Google Cloud 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.
-
-
Make sure that billing is enabled for your Google Cloud project.
-
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 -
Create local authentication credentials for your user account:
gcloud auth application-default login
-
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_IDENTIFIER" --role=ROLE
- Replace
PROJECT_ID
with your project ID. -
Replace
USER_IDENTIFIER
with the identifier for your user account. For example,user:myemail@example.com
. - Replace
ROLE
with each individual role.
- Replace
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:PROJECT_NUMBER-compute@developer.gserviceaccount.com" --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 commandegcloud projects describe
. - Remplacez
SERVICE_ACCOUNT_ROLE
par chaque rôle individuel.
-
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. - 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.
gcloud storage buckets create gs://BUCKET_NAME --default-storage-class STANDARD --location US
-
Set the storage class to
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 :
- 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
. - Analyser chaque ligne en mots
- 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.
Créez un répertoire pour votre module Go à l'emplacement de votre choix :
mkdir wordcount
cd wordcount
Créez un module Go. Pour cet exemple, utilisez
example/dataflow
comme chemin de module.go mod init example/dataflow
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épertoirewordcount
que vous avez créé.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
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.
Depuis le terminal, créez et exécutez le pipeline localement :
go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \ --output outputs
Affichez les résultats :
more outputs*
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.
Dans l'éditeur de votre choix, ouvrez le fichier
wordcount.go
.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]() }
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) }
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) }
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) }
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é.
Créez et exécutez le pipeline
wordcount
modifié :go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \ --output outputs
Affichez les résultats du pipeline modifié. Tous les mots doivent être en minuscules.
more outputs*
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.
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.
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.
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.
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é.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.
- In the Google Cloud console, go to the Cloud Storage Buckets page.
- Click the checkbox for the bucket that you want to delete.
- To delete the bucket, click Delete, and then follow the instructions.
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:PROJECT_NUMBER-compute@developer.gserviceaccount.com \ --role=SERVICE_ACCOUNT_ROLE
-
Optional: Revoke the authentication credentials that you created, and delete the local credential file.
gcloud auth application-default revoke
-
Optional: Revoke credentials from the gcloud CLI.
gcloud auth revoke