Présentation

Les workflows sont une notion courante dans l'analyse de données ; ils impliquent l'ingestion, la transformation et l'analyse de données pour déduire les informations significatives. Dans Google Cloud Platform (GCP), l'hébergement des workflows s'effectue dans Cloud Composer, version hébergée de l'outil de workflow Open Source communément utilisé Apache Airflow.

Dans cet atelier, vous allez configurer un environnement Cloud Composer dans la console GCP. Vous allez ensuite utiliser Cloud Composer pour mettre en place un workflow simple qui vérifie l'existence d'un fichier de données, crée un cluster Cloud Dataproc, exécute un job de décompte de mots Apache Hadoop sur le cluster Cloud Dataproc, puis supprime ce cluster.

Objectifs de l'atelier

• Créer l'environnement Cloud Composer dans la console GCP

• Afficher et exécuter le graphe orienté acyclique (DAG, Directed Acyclic Graph) dans l'interface Web Airflow

• Afficher les résultats du job de décompte dans l'espace de stockage

Préparation

Mettre en place l'atelier

Pour chaque atelier, nous vous attribuons un nouveau projet Google Cloud et un nouvel ensemble de ressources pour une durée déterminée, sans frais.

1. Connectez-vous à Qwiklabs dans une fenêtre de navigation privée.

2. Vérifiez le temps imparti pour l'atelier (par exemple : 01:15:00) : vous devez pouvoir le terminer dans ce délai.
   Une fois l'atelier lancé, vous ne pouvez pas le mettre en pause. Si nécessaire, vous pourrez le redémarrer, mais vous devrez tout reprendre depuis le début.

3. Lorsque vous êtes prêt, cliquez sur Démarrer l'atelier.

4. Notez vos identifiants pour l'atelier (Nom d'utilisateur et Mot de passe). Ils vous serviront à vous connecter à Google Cloud Console.

5. Cliquez sur Ouvrir la console Google.

6. Cliquez sur Utiliser un autre compte, puis copiez-collez les identifiants de cet atelier lorsque vous y êtes invité.
   Si vous utilisez d'autres identifiants, des messages d'erreur s'afficheront ou des frais seront appliqués.

7. Acceptez les conditions d'utilisation et ignorez la page concernant les ressources de récupération des données.

Activer Cloud Shell

Cloud Shell est une machine virtuelle qui contient des outils pour les développeurs. Elle comprend un répertoire d'accueil persistant de 5 Go et s'exécute sur Google Cloud. Google Cloud Shell vous permet d'accéder via une ligne de commande à vos ressources Google Cloud. gcloud est l'outil de ligne de commande associé à Google Cloud. Il est préinstallé sur Cloud Shell et permet la saisie semi-automatique via la touche Tabulation.

1. Dans Google Cloud Console, dans le volet de navigation, cliquez sur Activer Cloud Shell ().

2. Cliquez sur Continuer.
   Le provisionnement et la connexion à l'environnement prennent quelques instants. Une fois connecté, vous êtes en principe authentifié, et le projet est défini sur votre ID_PROJET. Exemple :

Exemples de commandes

• Afficher le nom du compte actif :

gcloud auth list

(Résultat)

Credentialed accounts: - <myaccount>@<mydomain>.com (active)

(Exemple de résultat)

Credentialed accounts: - google1623327_student@qwiklabs.net

• Afficher l'ID du projet :

gcloud config list project

(Résultat)

[core] project = <ID_Projet>

(Exemple de résultat)

[core] project = qwiklabs-gcp-44776a13dea667a6

Vérifier les autorisations du projet

Avant de commencer à travailler dans Google Cloud, vous devez vous assurer de disposer des autorisations adéquates pour votre projet dans IAM (Identity and Access Management).

1. Dans la console Google Cloud, accédez au menu de navigation (), puis cliquez sur IAM et administration > IAM.

2. Vérifiez que le compte de service Compute par défaut {project-number}-compute@developer.gserviceaccount.com existe et qu'il est associé au rôle Éditeur. Le préfixe du compte correspond au numéro du projet, disponible sur cette page : Menu de navigation > Présentation Cloud.

Si le compte n'est pas disponible dans IAM ou n'est pas associé au rôle Éditeur, procédez comme suit pour lui attribuer le rôle approprié.

1. Dans la console Google Cloud, accédez au menu de navigation et cliquez sur Présentation Cloud.

2. Sur la carte Informations sur le projet, copiez le numéro du projet.

3. Dans le menu de navigation, cliquez sur IAM et administration > IAM.

4. En haut de la page IAM, cliquez sur Ajouter.

5. Dans le champ Nouveaux comptes principaux, saisissez :

{project-number}-compute@developer.gserviceaccount.com

Remplacez {project-number} par le numéro de votre projet.

6. Dans le champ Sélectionnez un rôle, sélectionnez De base (ou Projet) > Éditeur.

7. Cliquez sur Enregistrer.

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

Pour vous assurer que vous avez bien accès aux API nécessaires, redémarrez la connexion à l'API Kubernetes Engine.

1. Dans la console Google Cloud, recherchez et enregistrez le numéro du projet dans le champ Informations sur le projet.

2. Dans la console, saisissez API Kubernetes Engine dans la barre de recherche en haut de l'écran. Cliquez sur API Kubernetes Engine dans les résultats.

3. Cliquez sur Gérer.

4. Cliquez sur Désactiver l'API.

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

   • De même, à l'invite Voulez-vous désactiver l'API Kubernetes Engine et les API qui en dépendent ?, cliquez sur Désactiver.

5. Cliquez sur Activer.

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

Tâche 2 : Vérifier que l'API Cloud Composer est activée

Redémarrez la connexion à l'API Cloud Composer. Lors de l'étape précédente, le redémarrage de l'API Kubernetes Engine a forcé la désactivation de l'API Cloud Composer.

1. Dans la console Google Cloud, saisissez API Cloud Composer dans la barre de recherche en haut de l'écran. Cliquez sur le résultat qui s'affiche.

2. Cliquez sur Activer.

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

Tâche 3 : Créer un environnement Cloud Composer

Dans cette section, vous allez créer un environnement Cloud Composer.

Remarque : Avant de poursuivre, assurez-vous d'avoir bien effectué les étapes précédentes de sorte que les API requises soient activées. Si ce n'est pas le cas, veuillez effectuer ces étapes. Sinon, la création de l'environnement Cloud Composer échouera.

1. Dans la barre de titre de la console Google Cloud, saisissez cloud composer dans le champ Recherche, puis cliquez sur Composer.

2. Cliquez sur Créer un environnement et sélectionnez Composer 3.

3. Définissez les éléments suivants pour votre environnement :

Propriété	Valeur
Nom	highcpu
Emplacement
Version de l'image	composer-3-airflow-n.n.n-build.n (Remarque : sélectionnez la dernière version disponible de l'image)

Conservez les valeurs par défaut de tous les autres paramètres.

4. Sous Ressources de l'environnement, sélectionnez Petit.

5. Cliquez sur Créer.

Le processus de création de l'environnement est terminé lorsqu'une coche verte apparaît à gauche du nom de l'environnement sur la page "Environnements" de la console GCP.

La procédure de configuration de l'environnement peut prendre entre 10 et 20 minutes. Poursuivez l'atelier pendant que l'environnement se configure.

Cliquez sur Vérifier ma progression pour valider l'objectif.

Créer un environnement Cloud Composer

Créer un bucket Cloud Storage

Créez un bucket Cloud Storage dans votre projet. Ce bucket servira de sortie pour le job Hadoop fourni par Dataproc.

1. Accédez au menu de navigation > Cloud Storage > Buckets, puis cliquez sur + Créer.

2. Attribuez au bucket un nom unique universel, puis cliquez sur Créer. Si le message L'accès public sera bloqué s'affiche, cliquez sur Confirmer.

Notez bien le nom du bucket Cloud Storage, car vous l'utiliserez comme variable Airflow plus tard dans l'atelier.

Cliquez sur Vérifier ma progression pour valider l'objectif.

Créer un bucket Cloud Storage

Tâche 4 : Airflow et concepts fondamentaux

En attendant la création de votre environnement Composer, découvrez quelques-uns des termes associés à Airflow.

Airflow est une plate-forme qui permet de créer, de planifier et de surveiller vos workflows de manière programmatique.

Avec Airflow, vous pouvez créer des workflows en tant que graphes de tâches orientés acycliques. Le programmeur Airflow exécute vos tâches sur un tableau de nœuds de calcul tout en suivant les dépendances spécifiées.

Concepts fondamentaux

DAG : un graphe orienté acyclique (Directed Acyclic Graph) est une collection regroupant toutes les tâches à exécuter et organisée de manière à refléter leurs relations et leurs dépendances.

Opérateur : description d'une tâche unique, généralement atomique. Par exemple, l'opérateur BashOperator permet d'exécuter une commande Bash.

Tâche : instance paramétrée d'un opérateur. Cette tâche devient un nœud dans le DAG.

Instance de tâche : exécution spécifique d'une tâche. Elle se compose d'un DAG, d'une tâche et d'une date/heure d'exécution. Un état indicatif lui est attribué : running (exécution en cours), success (réussite), failed (échec), skipped (ignorée), ...

Pour en savoir plus, lisez la documentation sur les concepts.

Tâche 5 : Définir le workflow

Étudions maintenant le workflow que vous allez utiliser. Les workflows Cloud Composer sont constitués de DAG (graphes orientés acycliques). Les DAG sont définis dans des fichiers Python standards placés dans le dossier DAG_FOLDER d'Airflow. Airflow exécute le code de chaque fichier pour créer dynamiquement les objets DAG. Vous pouvez avoir autant de DAG que vous voulez, chacun décrivant un nombre arbitraire de tâches. En général, un DAG correspond à un workflow logique unique.

Le code du workflow hadoop_tutorial.py, que nous appellerons également le "DAG", est reproduit ci-dessous :

"""Example Airflow DAG that creates a Cloud Dataproc cluster, runs the Hadoop wordcount example, and deletes the cluster. This DAG relies on three Airflow variables https://airflow.apache.org/concepts.html#variables * gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster. * gce_zone - Google Compute Engine zone where Cloud Dataproc cluster should be created. * gcs_bucket - Google Cloud Storage bucket to used as output for the Hadoop jobs from Dataproc. See https://cloud.google.com/storage/docs/creating-buckets for creating a bucket. """ import datetime import os from airflow import models from airflow.contrib.operators import dataproc_operator from airflow.utils import trigger_rule # Output file for Cloud Dataproc job. output_file = os.path.join( models.Variable.get('gcs_bucket'), 'wordcount', datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep # Path to Hadoop wordcount example available on every Dataproc cluster. WORDCOUNT_JAR = ( 'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar' ) # Arguments to pass to Cloud Dataproc job. wordcount_args = ['wordcount', 'gs://pub/shakespeare/rose.txt', output_file] yesterday = datetime.datetime.combine( datetime.datetime.today() - datetime.timedelta(1), datetime.datetime.min.time()) default_dag_args = { # Setting start date as yesterday starts the DAG immediately when it is # detected in the Cloud Storage bucket. 'start_date': yesterday, # To email on failure or retry set 'email' arg to your email and enable # emailing here. 'email_on_failure': False, 'email_on_retry': False, # If a task fails, retry it once after waiting at least 5 minutes 'retries': 1, 'retry_delay': datetime.timedelta(minutes=5), 'project_id': models.Variable.get('gcp_project') } with models.DAG( 'composer_sample_quickstart', # Continue to run DAG once per day schedule_interval=datetime.timedelta(days=1), default_args=default_dag_args) as dag: # Create a Cloud Dataproc cluster. create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator( task_id='create_dataproc_cluster', # Give the cluster a unique name by appending the date scheduled. # See https://airflow.apache.org/code.html#default-variables cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', num_workers=2, region=models.Variable.get('gce_region'), zone=models.Variable.get('gce_zone'), image_version='2.0', master_machine_type='n1-standard-2', worker_machine_type='n1-standard-2') # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster # master node. run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator( task_id='run_dataproc_hadoop', region=models.Variable.get('gce_region'), main_jar=WORDCOUNT_JAR, cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', arguments=wordcount_args) # Delete Cloud Dataproc cluster. delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator( task_id='delete_dataproc_cluster', region=models.Variable.get('gce_region'), cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}', # Setting trigger_rule to ALL_DONE causes the cluster to be deleted # even if the Dataproc job fails. trigger_rule=trigger_rule.TriggerRule.ALL_DONE) # Define DAG dependencies. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Pour orchestrer les trois tâches de workflow, le DAG importe les opérateurs suivants :

1. DataprocClusterCreateOperator : crée un cluster Cloud Dataproc.
2. DataProcHadoopOperator : soumet un job de décompte de mots Hadoop et code le résultat dans un bucket Cloud Storage.
3. DataprocClusterDeleteOperator : supprime le cluster pour éviter que des frais Compute Engine ne continuent d'être facturés.

Les tâches sont exécutées de manière séquentielle, comme vous pouvez le voir dans cette section du fichier :

# Define DAG dependencies. create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Ce DAG, nommé quickstart, s'exécute une fois par jour.

with models.DAG( 'composer_sample_quickstart', # Continue to run DAG once per day schedule_interval=datetime.timedelta(days=1), default_args=default_dag_args) as dag:

La valeur start_date transmise à default_dag_args étant définie sur yesterday, Cloud Composer planifie le démarrage du workflow immédiatement après l'importation du DAG.

Tâche 6 : Afficher les informations concernant l'environnement

1. Retournez à Composer pour vérifier l'état de votre environnement.

2. Une fois l'environnement créé, cliquez sur son nom (highcpu) pour en afficher les détails.

L'onglet Configuration de l'environnement contient des informations telles que l'URL de l'interface utilisateur Web d'Airflow, le cluster GKE et un lien vers le dossier des DAG, stocké dans votre bucket.

Remarque : Cloud Composer planifie uniquement les workflows dans le dossier /dags.

Tâche 7 : Utiliser l'interface utilisateur d'Airflow

Pour accéder à l'interface Web d'Airflow à partir de la console GCP :

1. Retournez à la page Environnements.
2. Dans la colonne Serveur Web Airflow associée à l'environnement, cliquez sur Airflow.
3. Cliquez sur les identifiants qui vous ont été attribués pour cet atelier.
4. L'interface Web Airflow s'ouvre dans une nouvelle fenêtre du navigateur.

Tâche 8 : Définir les variables Airflow

Les variables Airflow sont un concept spécifique à Airflow et sont différentes des variables d'environnement.

1. Dans la barre de menu de l'interface d'Airflow, sélectionnez Administrateur > Variables.

2. Cliquez sur l'icône + pour ajouter un enregistrement.

3. Créez les variables Airflow gcp_project, gcs_bucket et gce_zone suivantes :

Clé	Valeur	Détails
gcp_project		Le projet Google Cloud Platform que vous utilisez dans cet atelier.
gcs_bucket	gs://<my-bucket>	Remplacez <my-bucket> par le nom du bucket Cloud Storage que vous avez créé précédemment. Ce bucket stocke le résultat des jobs Hadoop exécutées sur Dataproc.
gce_zone		Il s'agit de la zone Compute Engine dans laquelle votre cluster Cloud Dataproc sera créé.
gce_region		Il s'agit de la région Compute Engine dans laquelle votre cluster Cloud Dataproc sera créé.

4. Cliquez sur Enregistrer. Après avoir ajouté la première variable, répétez la même procédure pour la deuxième et la troisième variable. La table des variables doit se présenter comme suit lorsque vous avez terminé :

Tâche 9 : Importer le DAG dans Cloud Storage

Pour importer le DAG :

1. Dans Cloud Shell, exécutez la commande ci-dessous pour importer une copie du fichier hadoop_tutorial.py dans le bucket Cloud Storage qui a été créé automatiquement lors de la création de l'environnement.

2. Dans la commande suivante, remplacez <DAGs_folder_path> par le chemin d'accès au dossier des DAG :

gsutil cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py <DAGs_folder_path>

• Pour obtenir le chemin d'accès, accédez à Composer.
• Cliquez sur l'environnement que vous avez créé précédemment, puis sur l'onglet Configuration de l'environnement pour afficher les informations de l'environnement.
• Recherchez le dossier des DAG et copiez le chemin d'accès.

La commande modifiée permettant d'importer le fichier doit ressembler à la suivante :

gsutil cp gs://cloud-training/datawarehousing/lab_assets/hadoop_tutorial.py gs://{{{project_0.default_region|REGION}}}-highcpu-0682d8c0-bucket/dags

3. Une fois le fichier importé dans le répertoire des DAG, ouvrez le dossier dags dans le bucket. Le fichier apparaît dans l'onglet Objets de la page "Informations sur le bucket".

Lorsqu'un fichier DAG est ajouté au dossier des DAG, Cloud Composer ajoute le DAG à Airflow et le programme automatiquement. Les modifications sont appliquées au DAG après trois à cinq minutes.

L'état de la tâche du DAG composer_hadoop_tutorial s'affiche dans l'interface Web d'Airflow.

Cliquez sur Vérifier ma progression pour valider l'objectif.

Importer le DAG dans Cloud Storage

Explorer les exécutions du DAG

Lorsque vous importez le fichier DAG dans le dossier dags de Cloud Storage, il est analysé par Cloud Composer. Si aucune erreur n'est détectée, le nom du workflow apparaît dans la liste des DAG, et le workflow est placé en file d'attente en vue d'une exécution immédiate.

Assurez-vous que l'onglet des DAG de l'interface Web Airflow est actif. Ce processus peut prendre plusieurs minutes. Actualisez le navigateur pour afficher les informations les plus récentes.

1. Assurez-vous que l'onglet des DAG de l'interface Web Airflow est actif. Ce processus peut prendre plusieurs minutes. Actualisez le navigateur pour afficher les informations les plus récentes.

2. Dans Airflow, cliquez sur composer_hadoop_tutorial pour ouvrir la page d'informations du DAG. Cette page comprend plusieurs représentations des tâches et des dépendances du workflow.

3. Dans la barre d'outils, cliquez sur Graphique. Passez la souris sur le graphique de chaque tâche pour afficher l'état de celle-ci. Notez que la bordure autour de chaque tâche indique également son état (bordure verte = exécution en cours, rouge = échec, etc.).

4. Cliquez sur le lien "Actualiser" pour afficher les informations les plus récentes. En fonction de l'état d'évolution des processus, la couleur des bordures change.

Remarque : Si votre cluster Dataproc existe déjà, vous pouvez de nouveau exécuter le workflow pour atteindre l'état de réussite. Pour ce faire, cliquez sur le graphique "create_dataproc_cluster", choisissez Effacer pour réinitialiser les trois tâches, puis cliquez sur OK pour confirmer.

5. Lorsque create_dataproc_cluster passe à l'état "running" (en cours d'exécution), accédez au menu de navigation > Dataproc, puis cliquez sur les éléments suivants :

   • Clusters pour surveiller la création et la suppression des clusters. Le cluster créé par le workflow est éphémère : il est supprimé lors de la dernière tâche de ce workflow.
   • Jobs pour surveiller le job de décompte de mots Apache Hadoop. Cliquez sur l'ID de job pour afficher la sortie du journal associée au job.

6. Une fois que Dataproc passe à l'état "Running" (Exécution en cours), retournez à Airflow, puis cliquez sur Refresh (Actualiser) pour vérifier que le cluster est créé.

Une fois le processus run_dataproc_hadoop terminé, accédez au menu de navigation, sélectionnez Cloud Storage > Buckets, puis cliquez sur le nom de votre bucket pour afficher les résultats du décompte de mots dans le dossier wordcount.

7. Une fois toutes les étapes terminées dans le DAG, chacune présente une bordure vert foncé. De plus, le cluster Dataproc qui a été créé est désormais supprimé.

Félicitations !

Vous avez réussi à exécuter un workflow Cloud Composer.

Étapes suivantes

• Découvrez la présentation de Cloud Composer lors de la conférence NEXT 18 à San Francisco : https://www.youtube.com/watch?v=GeNFEtt-D4k
• Pour afficher la valeur d'une variable, exécutez les variables de la sous-commande CLI Airflow avec l'argument get ou utilisez l'interface Web Airflow.
• Pour en savoir plus sur l'interface Web Airflow, consultez la page Accéder à l'interface Web.

Terminer l'atelier

Une fois l'atelier terminé, cliquez sur End Lab (Terminer l'atelier). Qwiklabs supprime les ressources que vous avez utilisées, puis efface le compte.

Si vous le souhaitez, vous pouvez noter l'atelier. Sélectionnez le nombre d'étoiles correspondant à votre note, saisissez un commentaire, puis cliquez sur Submit (Envoyer).

Le nombre d'étoiles que vous pouvez attribuer à un atelier correspond à votre degré de satisfaction :

• 1 étoile = très mécontent(e)
• 2 étoiles = insatisfait(e)
• 3 étoiles = ni insatisfait(e), ni satisfait(e)
• 4 étoiles = satisfait(e)
• 5 étoiles = très satisfait(e)

Si vous ne souhaitez pas donner votre avis, vous pouvez fermer la boîte de dialogue.

Pour soumettre des commentaires, suggestions ou corrections, veuillez utiliser l'onglet Support (Assistance).

Copyright 2020 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.