GSP290

Présentation

Dataflow est un service Google Cloud qui fournit un traitement unifié des données par flux et par lot à grande échelle. Il repose sur le projet Apache Beam, qui est un modèle Open Source permettant de définir des pipelines de traitement parallèle des données par lot et par flux. En utilisant l'un des SDK Apache Beam Open Source, vous pouvez créer un programme qui définit le pipeline, puis fait appel à Dataflow pour l'exécuter.

Dans cet atelier, vous allez utiliser le SDK Apache Beam pour Python afin de créer et d'exécuter dans Dataflow un pipeline qui ingère des données dans BigQuery depuis Cloud Storage, puis transforme et enrichit ces données dans BigQuery.

Remarque : Pensez à ouvrir les fichiers Python et à bien lire les commentaires lorsque vous y êtes invité. Cela vous permettra de comprendre le fonctionnement du code.

Objectifs de l'atelier

Dans cet atelier, vous allez apprendre à :

• créer et exécuter des pipelines Dataflow (Python) pour ingérer des données depuis Cloud Storage dans BigQuery ;
• créer et exécuter des pipelines Dataflow (Python) pour transformer et enrichir des données dans BigQuery ;
• créer et exécuter des pipelines Dataflow (Python) pour effectuer une jointure de données dans BigQuery et écrire les résultats dans une nouvelle table.

Préparation

Avant de cliquer sur le bouton "Démarrer l'atelier"

Lisez ces instructions. Les ateliers sont minutés, et vous ne pouvez pas les mettre en pause. Le minuteur, qui démarre lorsque vous cliquez sur Démarrer l'atelier, indique combien de temps les ressources Google Cloud resteront accessibles.

Cet atelier pratique vous permet de suivre les activités dans un véritable environnement cloud, et non dans un environnement de simulation ou de démonstration. Des identifiants temporaires vous sont fournis pour vous permettre de vous connecter à Google Cloud le temps de l'atelier.

Pour réaliser cet atelier :

• Vous devez avoir accès à un navigateur Internet standard (nous vous recommandons d'utiliser Chrome).

Remarque : Ouvrez une fenêtre de navigateur en mode incognito (recommandé) ou de navigation privée pour effectuer cet atelier. Vous éviterez ainsi les conflits entre votre compte personnel et le compte temporaire de participant, qui pourraient entraîner des frais supplémentaires facturés sur votre compte personnel.

• Vous disposez d'un temps limité. N'oubliez pas qu'une fois l'atelier commencé, vous ne pouvez pas le mettre en pause.

Remarque : Utilisez uniquement le compte de participant pour cet atelier. Si vous utilisez un autre compte Google Cloud, des frais peuvent être facturés à ce compte.

Démarrer l'atelier et se connecter à la console Google Cloud

1. Cliquez sur le bouton Démarrer l'atelier. Si l'atelier est payant, une boîte de dialogue s'affiche pour vous permettre de sélectionner un mode de paiement. Sur la gauche, vous trouverez le panneau "Détails concernant l'atelier", qui contient les éléments suivants :

   • Le bouton "Ouvrir la console Google Cloud"
   • Le temps restant
   • Les identifiants temporaires que vous devez utiliser pour cet atelier
   • Des informations complémentaires vous permettant d'effectuer l'atelier

2. Cliquez sur Ouvrir la console Google Cloud (ou effectuez un clic droit et sélectionnez Ouvrir le lien dans la fenêtre de navigation privée si vous utilisez le navigateur Chrome).

   L'atelier lance les ressources, puis ouvre la page "Se connecter" dans un nouvel onglet.

   Conseil : Réorganisez les onglets dans des fenêtres distinctes, placées côte à côte.

   Remarque : Si la boîte de dialogue Sélectionner un compte s'affiche, cliquez sur Utiliser un autre compte.

3. Si nécessaire, copiez le nom d'utilisateur ci-dessous et collez-le dans la boîte de dialogue Se connecter.

   {{{user_0.username | "Username"}}}

   Vous trouverez également le nom d'utilisateur dans le panneau "Détails concernant l'atelier".

4. Cliquez sur Suivant.

5. Copiez le mot de passe ci-dessous et collez-le dans la boîte de dialogue Bienvenue.

   {{{user_0.password | "Password"}}}

   Vous trouverez également le mot de passe dans le panneau "Détails concernant l'atelier".

6. Cliquez sur Suivant.

   Important : Vous devez utiliser les identifiants fournis pour l'atelier. Ne saisissez pas ceux de votre compte Google Cloud. Remarque : Si vous utilisez votre propre compte Google Cloud pour cet atelier, des frais supplémentaires peuvent vous être facturés.

7. Accédez aux pages suivantes :

   • Acceptez les conditions d'utilisation.
   • N'ajoutez pas d'options de récupération ni d'authentification à deux facteurs (ce compte est temporaire).
   • Ne vous inscrivez pas à des essais sans frais.

Après quelques instants, la console Cloud s'ouvre dans cet onglet.

Remarque : Pour accéder aux produits et services Google Cloud, cliquez sur le menu de navigation ou saisissez le nom du service ou du produit dans le champ Recherche.

Activer Cloud Shell

Cloud Shell est une machine virtuelle qui contient de nombreux outils pour les développeurs. Elle comprend un répertoire d'accueil persistant de 5 Go et s'exécute sur Google Cloud. Cloud Shell vous permet d'accéder via une ligne de commande à vos ressources Google Cloud.

1. Cliquez sur Activer Cloud Shell  en haut de la console Google Cloud.

2. Passez les fenêtres suivantes :

   • Accédez à la fenêtre d'informations de Cloud Shell.
   • Autorisez Cloud Shell à utiliser vos identifiants pour effectuer des appels d'API Google Cloud.

Une fois connecté, vous êtes en principe authentifié et le projet est défini sur votre ID_PROJET : . Le résultat contient une ligne qui déclare l'ID_PROJET pour cette session :

Your Cloud Platform project in this session is set to {{{project_0.project_id | "PROJECT_ID"}}}

gcloud est l'outil de ligne de commande pour Google Cloud. Il est préinstallé sur Cloud Shell et permet la complétion par tabulation.

3. (Facultatif) Vous pouvez lister les noms des comptes actifs à l'aide de cette commande :

gcloud auth list

4. Cliquez sur Autoriser.

Résultat :

ACTIVE: * ACCOUNT: {{{user_0.username | "ACCOUNT"}}} To set the active account, run: $ gcloud config set account `ACCOUNT`

5. (Facultatif) Vous pouvez lister les ID de projet à l'aide de cette commande :

gcloud config list project

Résultat :

[core] project = {{{project_0.project_id | "PROJECT_ID"}}} Remarque : Pour consulter la documentation complète sur gcloud, dans Google Cloud, accédez au guide de présentation de la gcloud CLI.

Tâche 1 : Vérifier que l'API Dataflow est activée

Pour vous assurer que vous avez bien accès à l'API requise, redémarrez la connexion à l'API Dataflow.

Important : Même si l'API est actuellement activée, suivez les étapes 1 à 4 ci-dessous pour la désactiver puis la réactiver, afin de la redémarrer correctement.

1. Dans la barre de titre de la console Google Cloud, saisissez API Dataflow dans le champ Recherche, puis cliquez sur API Dataflow dans les résultats de recherche.

2. Cliquez sur Gérer.

3. Cliquez sur Désactiver l'API.

Si vous êtes invité à confirmer votre choix, cliquez sur Désactiver.

4. Cliquez sur Activer.

Une fois l'API réactivée, l'option Désactiver s'affiche sur la page.

Cliquez sur Vérifier ma progression pour valider la tâche exécutée.

Désactiver et réactiver l'API Dataflow

Tâche 2 : Télécharger le code de démarrage

Téléchargez les exemples Dataflow en Python que vous utiliserez dans cet atelier.

1. Exécutez la commande suivante dans Cloud Shell pour obtenir des exemples Dataflow en Python à partir du dépôt GitHub des services professionnels de Google Cloud :

gcloud storage cp -r gs://spls/gsp290/dataflow-python-examples .

2. Définissez une variable pour votre ID de projet.

export PROJECT={{{ project_0.project_id }}} gcloud config set project $PROJECT

Tâche 3 : Créer un bucket Cloud Storage et y copier des fichiers

Dans Cloud Shell, vous allez créer un bucket Cloud Storage, puis y copier des fichiers. Ces fichiers sont les exemples Dataflow en Python.

Créer un bucket Cloud Storage

• Toujours dans Cloud Shell, exécutez la commande de création de bucket pour créer un bucket régional dans la région au niveau de votre projet :

gcloud storage buckets create gs://$PROJECT --location={{{ project_0.default_region | REGION }}}

Cliquez sur Vérifier ma progression pour valider la tâche exécutée.

Créer un bucket Cloud Storage

Copier des fichiers dans votre bucket

• Dans Cloud Shell, exécutez la commande gsutil pour copier des fichiers dans le bucket Cloud Storage que vous venez de créer :

gcloud storage cp gs://spls/gsp290/data_files/usa_names.csv gs://$PROJECT/data_files/ gcloud storage cp gs://spls/gsp290/data_files/head_usa_names.csv gs://$PROJECT/data_files/

Cliquez sur Vérifier ma progression pour valider la tâche exécutée.

Copier des fichiers dans votre bucket

Tâche 4 : Créer un ensemble de données BigQuery

Vous allez créer un ensemble de données BigQuery, qui servira à charger vos tables dans BigQuery.

Dans Cloud Shell, créez l'ensemble de données nommé lake :

bq mk lake

Cliquez sur Vérifier ma progression pour valider la tâche exécutée.

Créer l'ensemble de données BigQuery nommé "lake"

Tâche 5 : Examiner et exécuter le pipeline d'ingestion de données

Dans cette tâche, vous allez examiner le code du pipeline pour comprendre son fonctionnement. Après cela, vous configurerez et exécuterez le pipeline.

Le pipeline d'ingestion de données ingère les données à partir de Cloud Storage dans la table BigQuery à l'aide d'une source TextIO et d'une destination BigQueryIO. Plus précisément, le pipeline effectue les actions suivantes :

• Ingérer les fichiers à partir de Cloud Storage
• Filtrer la ligne d'en-tête dans les fichiers
• Convertir les lignes lues en objets de dictionnaire
• Écrire les lignes dans BigQuery

Examiner le code Python du pipeline d'ingestion de données

Utilisez l'éditeur de code Cloud Shell pour examiner le code du pipeline.

1. Dans la barre de menu Cloud Shell, cliquez sur Ouvrir l'éditeur.

2. Accédez à dataflow_python_examples > dataflow_python_examples, puis ouvrez le fichier data_ingestion.py.

3. Lisez les commentaires expliquant les actions effectuées par le code.

Ce code insère dans une table BigQuery les données issues de fichiers stockés dans Cloud Storage.

4. Pour revenir dans Cloud Shell, cliquez sur Ouvrir le terminal.

Configurer le conteneur Docker pour les jobs Dataflow

1. Revenez dans votre session Cloud Shell pour configurer les bibliothèques Python requises.

Les jobs Dataflow de cet atelier nécessitent Python 3.8. Pour vous assurer d'utiliser la bonne version, exécutez les processus Dataflow dans un conteneur Docker Python 3.8.

2. Exécutez la commande suivante dans Cloud Shell pour démarrer un conteneur Python :

cd ~ docker run -it -e PROJECT=$PROJECT -v $(pwd)/dataflow-python-examples:/dataflow python:3.8 /bin/bash

Cette commande extrait un conteneur Docker avec la dernière version stable de Python 3.8, puis lance un shell de commande vous permettant d'exécuter les commandes suivantes dans votre conteneur. L'option -v fournit le code source en tant que volume pour le conteneur. Ainsi, nous pouvons le modifier dans l'éditeur Cloud Shell tout en gardant la possibilité d'y accéder dans le conteneur en cours d'exécution.

3. Dès que l'extraction du conteneur est terminée et que celui-ci est lancé dans Cloud Shell, exécutez la commande suivante pour installer apache-beam dans ce conteneur en cours d'exécution :

pip install apache-beam[gcp]==2.59.0

4. Ensuite, dans le conteneur en cours d'exécution dans Cloud Shell, accédez au répertoire auquel vous avez associé le code source :

cd dataflow/

5. Définissez l'ID du projet dans le conteneur :

export PROJECT={{{ project_0.project_id }}}

Exécuter le pipeline d'ingestion de données dans le cloud

1. Saisissez le code suivant pour exécuter le pipeline d'ingestion de données :

python dataflow_python_examples/data_ingestion.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

Ce code lance les nœuds de calcul requis et les arrête une fois l'exécution du pipeline terminée.

2. Dans la barre de titre de la console, saisissez Dataflow dans le champ Recherche, puis cliquez sur Dataflow dans les résultats de recherche.

Lorsque la page Dataflow s'ouvre, consultez l'état de votre job.

3. Cliquez sur le nom du job pour suivre sa progression.

Lorsque l'état du job indique Réussite, vous pouvez passer à l'étape suivante. Le démarrage, l'exécution et l'arrêt de ce pipeline d'ingestion prennent au total environ cinq minutes.

4. Accédez à BigQuery (Menu de navigation > BigQuery) pour vérifier que vos données ont bien été intégrées.

5. Cliquez sur le nom de votre projet pour afficher la table usa_names sous l'ensemble de données lake.

6. Cliquez sur la table, puis accédez à l'onglet Aperçu pour afficher des exemples de données de usa_names.

Remarque : Si la table usa_names n'apparaît pas, actualisez la page ou essayez d'afficher les tables dans l'UI classique de BigQuery.

Cliquez sur Vérifier ma progression pour valider la tâche exécutée.

Créer un pipeline d'ingestion de données

Tâche 6 : Examiner et exécuter le pipeline de transformation des données

Dans cette tâche, vous allez examiner le pipeline de transformation des données pour en comprendre le fonctionnement. Vous exécuterez ensuite le pipeline pour traiter les fichiers Cloud Storage et écrire le résultat dans BigQuery.

Le pipeline de transformation des données effectue aussi l'ingestion des données à partir de Cloud Storage dans la table BigQuery à l'aide d'une source TextIO et d'une destination BigQueryIO, mais il inclut des transformations de données supplémentaires. Plus précisément, le pipeline effectue les actions suivantes :

• Ingérer les fichiers à partir de Cloud Storage
• Convertir les lignes lues en objets de dictionnaire
• Transformer les données qui contiennent l'année dans un format que BigQuery reconnaît comme une date
• Écrire les lignes dans BigQuery

Examiner le code Python du pipeline de transformation des données

• Dans l'éditeur de code, ouvrez le fichier data_transformation.py.

Lisez les commentaires expliquant les actions effectuées par le code.

Exécuter le pipeline de transformation des données dans le cloud

1. Saisissez le code suivant pour exécuter le pipeline de transformation des données :

python dataflow_python_examples/data_transformation.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

2. Dans la barre de titre de la console Google Cloud, saisissez Dataflow dans le champ Recherche, puis cliquez sur Dataflow dans les résultats de recherche.

3. Cliquez sur le nom de ce job pour afficher son état.

Le démarrage, l'exécution et l'arrêt de ce pipeline Dataflow prennent au total environ cinq minutes.

4. Lorsque l'état du job indique Réussite dans l'écran Dataflow, accédez à BigQuery pour vérifier que vos données ont bien été intégrées.

Vous devez voir la table usa_names_transformed sous l'ensemble de données lake.

5. Cliquez sur la table, puis accédez à l'onglet Aperçu pour afficher des exemples de données de usa_names_transformed.

Remarque : Si la table usa_names_transformed n'apparaît pas, actualisez la page ou essayez d'afficher les tables dans l'UI classique de BigQuery.

Cliquez sur Vérifier ma progression pour valider la tâche exécutée.

Créer un pipeline de transformation de données

Tâche 7 : Examiner et exécuter le pipeline d'enrichissement des données

Vous allez maintenant créer un pipeline d'enrichissement des données qui effectue les opérations suivantes :

• Ingérer les fichiers à partir de Cloud Storage
• Filtrer la ligne d'en-tête dans les fichiers
• Convertir les lignes lues en objets de dictionnaire
• Envoyer les lignes dans BigQuery

Examiner et modifier le code Python du pipeline d'enrichissement des données

1. Dans l'éditeur de code, ouvrez le fichier data_enrichment.py.

2. Lisez les commentaires expliquant les actions effectuées par le code. Ce code insère les données dans BigQuery.

La ligne 83 se présente actuellement comme suit :

values = [x.decode('utf8') for x in csv_row]

3. Modifiez-la de sorte qu'elle ressemble à ceci :

values = [x for x in csv_row]

4. Lorsque vous avez terminé de modifier cette ligne, n'oubliez pas d'enregistrer le fichier mis à jour. Pour ce faire, cliquez sur le menu Fichier de l'éditeur de code, puis sur Enregistrer.

Exécuter le pipeline d'enrichissement des données

1. Saisissez le code suivant pour exécuter le pipeline d'enrichissement des données :

python dataflow_python_examples/data_enrichment.py \ --project=$PROJECT \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --input gs://$PROJECT/data_files/head_usa_names.csv \ --save_main_session

2. Sur la page Dataflow, cliquez sur votre job pour afficher son état.

Le démarrage, l'exécution et l'arrêt de ce pipeline Dataflow prennent au total environ cinq minutes.

3. Lorsque l'état du job indique Réussite dans l'écran Dataflow, cliquez sur le menu de navigation > BigQuery dans la console pour vérifier que vos données ont bien été intégrées.

Vous devez voir la table usa_names_enriched sous l'ensemble de données lake.

4. Cliquez sur la table, puis accédez à l'onglet Aperçu pour afficher des exemples de données de usa_names_enriched.

Remarque : Si la table usa_names_enriched n'apparaît pas, actualisez la page ou essayez d'afficher les tables dans l'UI classique de BigQuery.

Cliquez sur Vérifier ma progression pour valider la tâche exécutée.

Créer un pipeline d'enrichissement de données

Tâche 8 : Examiner et exécuter le pipeline entre le lac de données et le magasin

Vous allez maintenant créer un pipeline Dataflow qui lit les données de deux sources de données BigQuery et effectue leur jointure. Les opérations à réaliser sont les suivantes :

• Ingérer les fichiers de deux sources BigQuery
• Associer les deux sources de données
• Filtrer la ligne d'en-tête dans les fichiers
• Convertir les lignes lues en objets de dictionnaire
• Envoyer les lignes dans BigQuery

Exécuter le pipeline d'ingestion de données pour effectuer une jointure de données et écrire la table ainsi obtenue dans BigQuery

Vous allez d'abord examiner le code data_lake_to_mart.py pour comprendre son fonctionnement. Vous exécuterez ensuite le pipeline dans le cloud.

1. Dans l'éditeur de code, ouvrez le fichier data_lake_to_mart.py.

Lisez les commentaires expliquant les actions effectuées par le code. Ce code réalise la jointure de deux tables et écrit les résultats dans une nouvelle table dans BigQuery.

2. Saisissez le bloc de code suivant pour exécuter le pipeline :

python dataflow_python_examples/data_lake_to_mart.py \ --worker_disk_type="compute.googleapis.com/projects//zones//diskTypes/pd-ssd" \ --max_num_workers=4 \ --project=$PROJECT \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://$PROJECT/test \ --temp_location gs://$PROJECT/test \ --save_main_session \ --region={{{ project_0.default_region | REGION }}}

3. Dans la barre de titre de la console Google Cloud, saisissez Dataflow dans le champ Recherche, puis cliquez sur Dataflow dans les résultats de recherche.

4. Cliquez sur ce nouveau job pour afficher son état.

Le démarrage, l'exécution et l'arrêt de ce pipeline Dataflow prennent au total environ cinq minutes.

5. Lorsque l'état du job indique Réussite dans l'écran Dataflow, cliquez sur le menu de navigation > BigQuery pour vérifier que vos données ont bien été intégrées.

Vous devez voir la table orders_denormalized_sideinput sous l'ensemble de données lake.

6. Cliquez sur la table, puis accédez à l'onglet Aperçu pour afficher des exemples de données de orders_denormalized_sideinput.

Remarque : Si la table orders_denormalized_sideinput n'apparaît pas, actualisez la page ou essayez d'afficher les tables dans l'UI classique de BigQuery.

Cliquez sur Vérifier ma progression pour valider la tâche exécutée.

Créer un pipeline Dataflow du lac de données vers un magasin

Tester vos connaissances

Voici une question à choix multiples qui vous aidera à assimiler les concepts abordés lors de cet atelier. Répondez-y du mieux que vous le pouvez.

Félicitations !

Vous avez exécuté du code Python à l'aide de Dataflow pour ingérer des données depuis Cloud Storage vers BigQuery, puis les transformer et les enrichir dans BigQuery.

Étapes suivantes et informations supplémentaires

Vous souhaitez en savoir plus ? Consultez la documentation officielle sur :

• Dataflow
• BigQuery
• Consultez le guide de programmation d'Apache Beam pour en savoir plus sur les concepts avancés.
• Découvrez les ateliers suivants :
  • Prédire les achats des visiteurs avec un modèle de classification dans BQML
  • Cloud Composer : Copier des tables BigQuery à différents emplacements

Formations et certifications Google Cloud

Les formations et certifications Google Cloud vous aident à tirer pleinement parti des technologies Google Cloud. Nos cours portent sur les compétences techniques et les bonnes pratiques à suivre pour être rapidement opérationnel et poursuivre votre apprentissage. Nous proposons des formations pour tous les niveaux, à la demande, en salle et à distance, pour nous adapter aux emplois du temps de chacun. Les certifications vous permettent de valider et de démontrer vos compétences et votre expérience en matière de technologies Google Cloud.

Dernière mise à jour du manuel : 1er avril 2025

Dernier test de l'atelier : 1er avril 2025

Copyright 2025 Google LLC. Tous droits réservés. Google et le logo Google sont des marques de Google LLC. Tous les autres noms d'entreprises et de produits peuvent être des marques des entreprises auxquelles ils sont associés.