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

Dans ce guide de démarrage rapide, vous allez apprendre à utiliser le SDK Apache Beam pour Python afin de créer un programme qui définit un pipeline. Vous exécutez ensuite le pipeline à l'aide d'un exécuteur local direct ou basé sur le cloud, tel que 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).

Pour obtenir des conseils détaillés sur cette tâche directement dans la console Google Cloud, cliquez sur Visite guidée :

Visite guidée

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.

    • 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, BigQuery, Cloud Pub/Sub, Cloud Datastore, and Cloud Resource Manager APIs: 

    gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com 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:myemail@example.com.

    • Replace ROLE with each individual role.
    9.
  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.

    • 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, BigQuery, Cloud Pub/Sub, Cloud Datastore, and Cloud Resource Manager APIs: 

    gcloud services enable dataflow compute_component logging storage_component storage_api bigquery pubsub datastore.googleapis.com 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:myemail@example.com.

    • Replace ROLE with each individual role.
    16.

  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: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 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 document.

Configurer votre environnement

Dans cette section, utilisez l'invite de commande pour configurer un environnement virtuel Python isolé permettant d'exécuter votre projet de pipeline à l'aide de venv. Ce processus vous permet d'isoler les dépendances d'un projet des dépendances d'autres projets.

Si vous n'avez pas d'invite de commande à votre disposition, vous pouvez utiliser Cloud Shell. Le gestionnaire de packages pour Python 3 est déjà installé dans Cloud Shell, ce qui vous permet d'ignorer la création d'un environnement virtuel.

Pour installer Python, puis créer un environnement virtuel, procédez comme suit :

  1. Vérifiez que Python 3 et pip sont en cours d'exécution sur votre système : 
    python --version
python -m pip --version
  2. Si nécessaire, installez Python 3, puis configurez un environnement virtuel Python : suivez les instructions fournies dans les sections Installer Python et Configurer venv de la page Configurer un environnement de développement Python. Si vous utilisez Python 3.10 ou une version ultérieure, vous devez également activer l'exécuteur Dataflow v2. Pour utiliser l'exécuteur v1, utilisez Python 3.9 ou une version antérieure.

Une fois celui-ci terminé, vous pouvez désactiver l'environnement virtuel en exécutant la commande deactivate.

Obtenir le SDK Apache Beam

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.

Pour télécharger et installer le SDK Apache Beam, procédez comme suit :

  1. Vérifiez que vous êtes dans l'environnement virtuel Python que vous avez créé dans la section précédente. Assurez-vous que l'invite commence par <env_name>, où env_name est le nom de l'environnement virtuel.
  2. Installez la norme du package Python wheel : 
    pip install wheel
  3. Installez la dernière version du SDK Apache Beam pour Python :
    4. pip install 'apache-beam[gcp]'

    Sous Microsoft Windows, exécutez la commande suivante :

    pip install apache-beam[gcp]

    En fonction de la connexion, l'installation peut prendre un certain temps.

Exécuter le pipeline en local

Pour voir comment un pipeline s'exécute localement, utilisez un module Python prêt à l'emploi wordcount, qui est inclus dans le package apache_beam.

L'exemple de pipeline wordcount effectue les opérations suivantes :

  1. Utiliser un fichier texte comme entrée

    Ce fichier texte se trouve dans un bucket Cloud Storage dont le nom de ressource est 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 mettre le pipeline wordcount en local, procédez comme suit :

  1. Depuis votre terminal local, exécutez l'exemple wordcount : 
    python -m apache_beam.examples.wordcount \
  --output outputs
  2. Affichez le résultat du pipeline : 
    more outputs*
  3. Pour quitter, appuyez sur q.
L'exécution locale du pipeline vous permet de tester et de déboguer votre programme Apache Beam. Vous pouvez visualiser le code source wordcount.py sur le dépôt GitHub d'Apache Beam.

Exécuter le pipeline sur le service Dataflow

Dans cette section, exécutez l'exemple de pipeline wordcount du package apache_beam sur le service Dataflow. Cet exemple spécifie DataflowRunner comme paramètre pour --runner.
  • Exécutez le pipeline : 
    python -m apache_beam.examples.wordcount \
    --region DATAFLOW_REGION \
    --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner DataflowRunner \
    --project PROJECT_ID \
    --temp_location gs://BUCKET_NAME/tmp/

    Remplacez les éléments suivants :

    • DATAFLOW_REGION : région dans laquelle vous souhaitez déployer le job Dataflow (par exemple, europe-west1)

      L'option --region remplace la région par défaut définie dans le serveur de métadonnées, votre client local ou les variables d'environnement.

    • BUCKET_NAME : nom du bucket Cloud Storage que vous avez copié précédemment
    • PROJECT_ID : ID de projet Google Cloud que vous avez copié précédemment

Afficher les résultats

Lorsque vous exécutez un pipeline à l'aide de Dataflow, vos résultats sont stockés dans un bucket Cloud Storage. Dans cette section, vérifiez que le pipeline est en cours d'exécution à l'aide de la console Google Cloud ou du terminal local.

console Google Cloud

Pour afficher vos résultats dans la console Google Cloud, procédez comme suit :

  1. 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.

  2. Accédez à la page Buckets de Cloud Storage.

    Accéder à la page "Buckets"

  3. Dans la liste des buckets de votre projet, cliquez sur le bucket de stockage que vous avez créé précédemment.

    Dans le répertoire wordcount, les fichiers de sortie créés par votre tâche sont affichés.

Terminal local

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

    2. Remplacez BUCKET_NAME par le nom du bucket Cloud Storage utilisé dans le programme de pipeline.

  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*

Modifier le code du pipeline

Le pipeline wordcount dans les exemples précédents fait la distinction entre les mots en majuscules et en minuscules. Les étapes suivantes montrent comment effectuer des modifications afin que le pipeline wordcount ne soit pas sensible à la casse.
  1. Sur votre ordinateur local, téléchargez la dernière copie du code wordcount à partir du dépôt GitHub d'Apache Beam.
  2. Depuis le terminal local, exécutez le pipeline : 
    python wordcount.py --output outputs
  3. Afficher les résultats : 
    more outputs*
  4. Pour quitter, appuyez sur q.
  5. Dans l'éditeur de votre choix, ouvrez le fichier wordcount.py.
  6. Dans la fonction run, examinez les étapes du pipeline : 
    counts = (
        lines
        | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
        | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
        | 'GroupAndSum' >> beam.CombinePerKey(sum))

    Après l'élément split, les lignes sont divisées en mots sous forme de chaînes.

  7. Pour mettre les chaînes en minuscules, modifiez la ligne après split : 
    counts = (
        lines
        | 'Split' >> (beam.ParDo(WordExtractingDoFn()).with_output_types(str))
        | 'lowercase' >> beam.Map(str.lower)
        | 'PairWithOne' >> beam.Map(lambda x: (x, 1))
        | 'GroupAndSum' >> beam.CombinePerKey(sum))
    Cette modification mappe la fonction str.lower sur chaque mot. Cette ligne est équivalente à beam.Map(lambda word: str.lower(word)).
  8. Enregistrez le fichier et exécutez la tâche wordcount modifiée : 
    python wordcount.py --output outputs
  9. Affichez les résultats du pipeline modifié : 
    more outputs*
  10. Pour quitter, appuyez sur q.
  11. Exécuter le pipeline modifié sur le service Dataflow : 
    python wordcount.py \
    --region DATAFLOW_REGION \
    --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://BUCKET_NAME/results/outputs \
    --runner DataflowRunner \
    --project PROJECT_ID \
    --temp_location gs://BUCKET_NAME/tmp/

    Remplacez les éléments suivants :

    • DATAFLOW_REGION : région dans laquelle vous souhaitez déployer le job Dataflow
    • BUCKET_NAME : nom de votre bucket Cloud Storage.
    • PROJECT_ID : ID de votre projet Google Cloud.

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.

.
  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:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --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