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.
-
-
Vérifiez que la facturation est activée pour votre projet Google Cloud.
-
Activer les API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager :
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
-
Attribuez des rôles à votre compte Google. Exécutez la commande suivante une fois pour chacun des rôles IAM suivants :
roles/iam.serviceAccountUser
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE
- en remplaçant
PROJECT_ID
par l'ID de votre projet : - Remplacez
EMAIL_ADDRESS
par votre adresse e-mail. - Remplacez
ROLE
par chaque rôle individuel.
- en remplaçant
- 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.
-
-
Vérifiez que la facturation est activée pour votre projet Google Cloud.
-
Activer les API Dataflow, Compute Engine, Cloud Logging, Cloud Storage, Google Cloud Storage JSON, and Cloud Resource Manager :
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
-
Attribuez des rôles à votre compte Google. Exécutez la commande suivante une fois pour chacun des rôles IAM suivants :
roles/iam.serviceAccountUser
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE
- en remplaçant
PROJECT_ID
par l'ID de votre projet : - Remplacez
EMAIL_ADDRESS
par votre adresse e-mail. - Remplacez
ROLE
par chaque rôle individuel.
- en remplaçant
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.
-
Créez un bucket Cloud Storage et configurez-le comme suit :
-
Définissez la classe de stockage sur
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
-
Définissez la classe de stockage sur
- 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 :
- 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.
- Dans la console Google Cloud, accédez à la page Buckets de Cloud Storage.
- Cochez la case correspondant au bucket que vous souhaitez supprimer.
- Pour supprimer le bucket, cliquez sur Supprimer , puis suivez les 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
-
Facultatif : Révoquez les identifiants d'authentification que vous avez créés et supprimez le fichier d'identifiants local.
gcloud auth application-default revoke
-
Facultatif : Révoquez les identifiants de la CLI gcloud.
gcloud auth revoke