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 effectuer les opérations suivantes :

• Ingérer des données depuis Cloud Storage dans BigQuery
• Transformer et enrichir des données dans BigQuery
• 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 ci-dessous pour la désactiver puis la réactiver, afin de la redémarrer correctement.

gcloud services disable dataflow.googleapis.com --project {{{project_0.project_id|Project ID}}} --force gcloud services enable dataflow.googleapis.com --project {{{project_0.project_id|Project ID}}}

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.

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

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_0.project_id | BUCKET_NAME }}} --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_0.project_id | BUCKET_NAME }}}/data_files/ gcloud storage cp gs://spls/gsp290/data_files/head_usa_names.csv gs://{{{ project_0.project_id | BUCKET_NAME }}}/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

Activer Gemini Code Assist dans Cloud Shell IDE

Vous pouvez utiliser Gemini Code Assist dans un environnement de développement intégré (IDE) tel que Cloud Shell pour obtenir des conseils sur le code ou résoudre des problèmes liés à votre code. Avant de pouvoir utiliser Gemini Code Assist, vous devez l'activer.

1. Dans Cloud Shell, activez l'API Gemini pour Google Cloud à l'aide de la commande suivante :

gcloud services enable cloudaicompanion.googleapis.com

2. Cliquez sur Ouvrir l'éditeur dans la barre d'outils Cloud Shell.

Remarque : Pour ouvrir l'éditeur Cloud Shell, cliquez sur Ouvrir l'éditeur dans la barre d'outils Cloud Shell. Vous pouvez passer de Cloud Shell à l'éditeur de code et inversement en cliquant sur Ouvrir l'éditeur ou sur Ouvrir le terminal selon le cas.

3. Dans le volet de gauche, cliquez sur l'icône Paramètres. Dans la vue Paramètres, recherchez Gemini Code Assist.

4. Recherchez l'option Geminicodeassist: Enable et assurez-vous que la case correspondante est cochée. Fermez ensuite les Paramètres.

5. Cliquez sur Cloud Code - Aucun projet dans la barre d'état en bas de l'écran.

6. Autorisez le plug-in comme indiqué. Si aucun projet n'est sélectionné automatiquement, cliquez sur Sélectionner un projet Google Cloud, puis choisissez .

7. Vérifiez que votre projet Google Cloud () s'affiche dans le message d'état Cloud Code de la barre d'état.

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

Dans cette section, vous allez demander à Gemini Code Assist de donner plus d'informations sur le pipeline d'ingestion de données afin de le présenter à un nouveau membre de l'équipe.

1. Dans l'explorateur de fichiers de l'éditeur Cloud Shell, accédez à dataflow_python_examples > dataflow_python_examples > data_ingestion.py.

2. Ouvrez le fichier data_ingestion.py. Cette action active Gemini Code Assist, comme l'indique l'icône en haut à droite de l'éditeur.

3. Cliquez sur l'icône Gemini Code Assist : Actions intelligentes , puis sélectionnez Expliquer.

4. Gemini Code Assist ouvre un volet de chat avec le prompt prédéfini Explain this (Explique-moi ça). Dans la zone de texte intégrée au volet de discussion Code Assist, remplacez le prompt prérempli par le suivant, puis cliquez sur Envoyer :

Tu es ingénieur de données expert chez Cymbal AI. Un nouveau membre de l'équipe a besoin de se familiariser avec ce code de pipeline. Explique l'objectif et le fonctionnement du pipeline d'ingestion de données défini dans data_ingestion.py. Ton explication doit inclure les éléments suivants : 1. Un résumé de ce que fait le script. 2. La liste des composants clés, comme la classe DataIngestion et la fonction d'exécution. 3. Des informations expliquant comment le script utilise le pipeline Apache Beam pour lire, traiter et écrire des données. 4. Le rôle des arguments de la ligne de commande et la manière dont ils sont utilisés. 5. Une description du format des données d'entrée et du schéma de la table BigQuery de sortie. Ne modifie pas ce fichier avec les suggestions d'amélioration.

Ce code insère dans une table BigQuery les données issues de fichiers stockés dans Cloud Storage. L'explication détaillée du code du fichier data_ingestion.py s'affiche dans la discussion Gemini Code Assist.

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

Configurer le conteneur Docker pour les jobs Dataflow

Dans cette section, vous allez revenir à 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.

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

cd ~ docker run -it -e PROJECT={{{ project_0.project_id | PROJECT_ID }}} -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.

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

3. 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/

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_0.project_id | PROJECT_ID }}} \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --input gs://{{{ project_0.project_id | BUCKET_NAME }}}/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 cette section, vous allez demander à Gemini Code Assist de fournir des informations supplémentaires sur le pipeline de transformation des données pour aider davantage le nouveau membre de l'équipe.

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

2. Dans l'éditeur Cloud Shell, toujours dans le même répertoire, accédez au fichier data_transformation.py. Comme précédemment, notez l'icône en haut à droite de l'éditeur.

3. Cliquez sur l'icône Gemini Code Assist : Actions intelligentes , puis sélectionnez Expliquer.

4. Gemini Code Assist ouvre un volet de chat avec le prompt prédéfini Explain this (Explique-moi ça). Dans la zone de texte intégrée au volet de discussion Code Assist, remplacez le prompt prérempli par le suivant, puis cliquez sur Envoyer :

Tu es ingénieur de données expert chez Cymbal AI. Un nouveau membre de l'équipe a besoin de se familiariser avec ce code de pipeline. Explique l'objectif et le fonctionnement du pipeline de transformation des données défini dans data_transformation.py. Ton explication doit inclure les éléments suivants : 1. Un résumé de ce que fait le script, expliquant quelles sont ses différences avec un simple pipeline d'ingestion. 2. La liste des composants clés, comme la classe DataTransformation et la fonction d'exécution. 3. Des informations expliquant comment le script utilise le pipeline Apache Beam pour lire les données d'un fichier, les transformer et les écrire dans une table BigQuery. 4. Une description de la manière dont le script gère le schéma BigQuery en lisant depuis un fichier JSON. 5. Une présentation de la logique de transformation des données avec la méthode parse_method, notamment de la manière dont elle convertit une année en type DATE. 6. Le rôle des arguments de la ligne de commande et la manière dont ils sont utilisés. Ne modifie pas ce fichier avec les suggestions d'amélioration.

L'explication du code dans le fichier data_transformation.py s'affiche dans la discussion Gemini Code Assist.

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

1. Saisissez la commande suivante dans le terminal Cloud Shell pour exécuter le pipeline de transformation des données :

python dataflow_python_examples/data_transformation.py \ --project={{{ project_0.project_id | PROJECT_ID }}} \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --input gs://{{{ project_0.project_id | BUCKET_NAME }}}/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 des 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

Dans cette section, vous allez exploiter les fonctionnalités optimisées par l'IA de Gemini Code Assist pour examiner et modifier le code Python du pipeline d'enrichissement des données.

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

2. Dans l'éditeur Cloud Shell, toujours dans le même répertoire, accédez à data_enrichment.py. Comme précédemment, notez l'icône en haut à droite de l'éditeur.

3. Cliquez sur l'icône Gemini Code Assist : Actions intelligentes  dans la barre d'outils.

4. Pour mettre à jour le code à la ligne 83, collez le prompt suivant dans le champ de texte intégré de Gemini Code Assist qui s'ouvre dans la barre d'outils.

In the data_enrichment.py file, update line 83 by replacing x.decode('utf8') with x.

5. Pour demander à Gemini Code Assist de modifier le code en conséquence, appuyez sur ENTRÉE.

6. Dans la vue Diff de Gemini, cliquez sur Accepter lorsque vous y êtes invité.

La ligne 83 modifiée du fichier data_enrichment.py se présente maintenant comme suit :

values = [x for x in csv_row]

7. 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 la commande suivante dans le terminal Cloud Shell pour exécuter le pipeline d'enrichissement des données :

python dataflow_python_examples/data_enrichment.py \ --project={{{ project_0.project_id | PROJECT_ID }}} \ --region={{{ project_0.default_region | REGION }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --input gs://{{{ project_0.project_id | BUCKET_NAME }}}/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 des 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.

Remarque : Pour en savoir plus sur le pipeline du lac de données vers un magasin, vous pouvez à nouveau demander à Gemini Code Assist d'expliquer le code, comme vous l'avez fait pour ses fonctionnalités optimisées par l'IA dans les tâches 5 et 6.

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_0.project_id | PROJECT_ID }}} \ --runner=DataflowRunner \ --machine_type=e2-standard-2 \ --staging_location=gs://{{{ project_0.project_id | BUCKET_NAME }}}/test \ --temp_location gs://{{{ project_0.project_id | BUCKET_NAME }}}/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 en utilisant Dataflow avec l'aide de Gemini Code Assist pour ingérer des données depuis Cloud Storage dans 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 septembre 2025

Dernier test de l'atelier : 1er septembre 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.