Exécuter une tâche Hadoop de décompte de mots sur un cluster Dataproc

1. Introduction

Les workflows sont un cas d'utilisation courant en analytique des données. Ils impliquent l'ingestion, la transformation et l'analyse de données pour y trouver les informations pertinentes. Dans Google Cloud Platform, Cloud Composer est l'outil d'orchestration des workflows. Il s'agit d'une version hébergée de l'outil de workflow Open Source populaire Apache Airflow. Dans cet atelier, vous allez utiliser Cloud Composer pour créer un workflow simple qui crée un cluster Cloud Dataproc, l'analyse à l'aide de Cloud Dataproc et d'Apache Hadoop, puis le supprime.

Qu'est-ce que Cloud Composer ?

Cloud Composer est un service d'orchestration de workflows entièrement géré qui vous permet de créer, de programmer et de surveiller vos pipelines dans des clouds ou des centres de données sur site. Basé sur le projet Open Source populaire Apache Airflow et sur le langage de programmation Python, Cloud Composer est facile à utiliser et permet le libre choix du fournisseur.

En utilisant Cloud Composer au lieu d'une instance locale d'Apache Airflow, les utilisateurs peuvent bénéficier du meilleur d'Airflow sans coûts d'installation ni de gestion.

Pour en savoir plus sur Cloud Composer, consultez la documentation.

Qu'est-ce qu'Apache Airflow ?

Apache Airflow est un outil Open Source qui permet de créer, de planifier et de surveiller des workflows de manière automatisée. Vous allez rencontrer plusieurs termes clés concernant Airflow tout au long de l'atelier :

• DAG (Directed Acyclic Graph) : un DAG est un ensemble de tâches organisées que vous souhaitez planifier et exécuter. Les DAG, également appelés workflows, sont définis dans des fichiers Python standards.
• Opérateur : un opérateur décrit une seule tâche dans un workflow.

Qu'est-ce que Cloud Dataproc ?

Cloud Dataproc est le service entièrement géré Apache Spark et Apache Hadoop de Google Cloud Platform. Cloud Dataproc s'intègre facilement à d'autres services GCP. Vous bénéficiez ainsi d'une plate-forme performante et complète pour vos tâches de traitement des données, d'analyse et de machine learning.

Pour plus d'informations sur Cloud Dataproc, consultez la documentation.

Objectifs de l'atelier

Cet atelier de programmation vous explique comment créer et exécuter dans Cloud Composer un workflow Apache Airflow qui effectue les tâches suivantes:

• crée un cluster Cloud Dataproc ;
• Exécute une tâche de décompte de mots Apache Hadoop sur le cluster et génère ses résultats dans Cloud Storage
• Suppression du cluster

Points abordés

• Créer et exécuter un workflow Apache Airflow dans Cloud Composer
• Utiliser Cloud Composer et Cloud Dataproc pour exécuter une analyse sur un ensemble de données
• Comment accéder à l'environnement Cloud Composer via la console Google Cloud Platform, le SDK Cloud et l'interface Web Airflow

Prérequis

• Compte GCP
• Connaissances de base de la CLI
• Connaissances de base sur Python

2. Configurer GCP

Créer le projet

Sélectionnez ou créez un projet Google Cloud Platform.

Notez l'ID de votre projet, car vous en aurez besoin lors des étapes suivantes.

Si vous créez un projet, l'ID du projet s'affiche juste en dessous du nom du projet sur la page de création.
Si vous avez déjà créé un projet, vous trouverez son ID sur la page d'accueil de la console, dans sa fiche d'informations

Remarque: Si vous ne prévoyez pas de conserver les ressources créées dans ce tutoriel, créez un projet au lieu de sélectionner un projet existant. Une fois que vous avez terminé, vous pouvez supprimer le projet en retirant toutes les ressources associées au projet et au tutoriel.

Activer les API

Activez les API Cloud Composer, Cloud Dataproc et Cloud Storage.Une fois activées, vous pouvez ignorer le bouton "Accéder aux identifiants" et passer à l'étape suivante du tutoriel.

Pour plus d'informations sur l'activation et la désactivation des API, consultez la documentation

Créer un environnement Composer

Créez un environnement Cloud Composer avec la configuration suivante: Nom: my-composer-environment Emplacement: us-central1 Zone : us-central1-a Vous pouvez conserver les valeurs par défaut de toutes les autres configurations. Cliquez sur "Créer" en bas de la page.

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 création prend quelques minutes. Vous pouvez quitter cette page pendant la création de l'environnement.

Créer un bucket Cloud Storage

Dans votre projet, créez un bucket Cloud Storage avec la configuration suivante : Nom: <ID de votre projet> Classe de stockage par défaut: multirégional Location (Pays) : United States (États-Unis) Modèle de contrôle d'accès: précis Appuyez sur "Créer" lorsque vous êtes prêt

3. Configurer Apache Airflow

Si la création de votre environnement Composer n'est pas terminée, passez à l'étape "Exemple de workflow" et revenez à cette partie après

Afficher des informations sur l'environnement Composer

Dans la console GCP, ouvrez la page Environnements.

Cliquez sur le nom de l'environnement pour en afficher les détails.

La page Détails de l'environnement contient des informations telles que l'URL de l'interface Web d'Airflow, l'ID du cluster Google Kubernetes Engine, le nom du bucket Cloud Storage et le chemin d'accès au dossier /dags.

Dans Airflow, un DAG (graphe orienté acyclique) est un ensemble de tâches organisées que vous souhaitez planifier et exécuter. Les DAG, également appelés workflows, sont définis dans des fichiers Python standards. Cloud Composer ne programme que les DAG contenus dans le dossier /dags. Le dossier /dags se trouve dans le bucket Cloud Storage que Cloud Composer crée automatiquement lorsque vous créez votre environnement.

Définir des variables d'environnement Apache Airflow

Les variables Apache Airflow sont un concept spécifique à Airflow, et se distinguent des variables d'environnement. Dans cette étape, vous allez définir les trois variables Airflow suivantes : gcp_project, gcs_bucket et gce_zone.

Utiliser gcloud pour définir des variables

Commencez par ouvrir Cloud Shell, sur lequel le SDK Cloud est déjà installé.

Définissez la variable d'environnement COMPOSER_INSTANCE sur le nom de votre environnement Composer.

COMPOSER_INSTANCE=my-composer-environment

Pour définir des variables Airflow à l'aide de l'outil de ligne de commande gcloud, exécutez la commande gcloud composer environments run avec la sous-commande variables. Cette commande gcloud composer exécute la sous-commande CLI Airflow variables. La sous-commande transmet les arguments à l'outil de ligne de commande gcloud.

Vous exécuterez cette commande trois fois, en remplaçant les variables par celles qui sont pertinentes pour votre projet.

Définissez gcp_project à l'aide de la commande suivante, en remplaçant <your-project-id> par l'ID du projet que vous avez noté à l'étape 2.

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gcp_project <your-project-id>

Votre résultat devrait se présenter comme suit :

kubeconfig entry generated for us-central1-my-composer-env-123abc-gke.
Executing within the following Kubernetes cluster namespace: composer-1-10-0-airflow-1-10-2-123abc
[2020-04-17 20:42:49,713] {settings.py:176} INFO - settings.configure_orm(): Using pool settings. pool_size=5, pool_recycle=1800, pid=449
[2020-04-17 20:42:50,123] {default_celery.py:90} WARNING - You have configured a result_backend of redis://airflow-redis-service.default.svc.cluste
r.local:6379/0, it is highly recommended to use an alternative result_backend (i.e. a database).
[2020-04-17 20:42:50,127] {__init__.py:51} INFO - Using executor CeleryExecutor
[2020-04-17 20:42:50,433] {app.py:52} WARNING - Using default Composer Environment Variables. Overrides have not been applied.
[2020-04-17 20:42:50,440] {configuration.py:522} INFO - Reading the config from /etc/airflow/airflow.cfg
[2020-04-17 20:42:50,452] {configuration.py:522} INFO - Reading the config from /etc/airflow/airflow.cfg

Définissez gcs_bucket à l'aide de la commande suivante, en remplaçant <your-bucket-name> par l'ID de bucket que vous avez noté à l'étape 2. Si vous avez suivi notre recommandation, le nom de votre bucket est identique à votre ID de projet. Le résultat sera semblable à celui de la commande précédente.

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gcs_bucket gs://<your-bucket-name>

Définissez gce_zone à l'aide de la commande suivante. Le résultat sera semblable à celui des commandes précédentes.

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --set gce_zone us-central1-a

(Facultatif) Utiliser gcloud pour afficher une variable

Pour afficher la valeur d'une variable, exécutez la sous-commande CLI Airflow variables avec l'argument get ou utilisez l'interface utilisateur Airflow.

Exemple :

gcloud composer environments run ${COMPOSER_INSTANCE} \
    --location us-central1 variables -- --get gcs_bucket

Vous pouvez le faire avec l'une des trois variables que vous venez de définir : gcp_project, gcs_bucket et gce_zone.

4. Exemple de workflow

Examinons le code du DAG que nous utiliserons à l'étape 5. Si vous n'avez pas encore besoin de télécharger des fichiers, suivez la procédure décrite ici.

Il y a beaucoup de choses à décompresser ici, alors analysons-les un peu.

from airflow import models
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rule

Nous commençons par quelques importations Airflow :

• airflow.models : permet d'accéder aux données de la base de données Airflow et de les créer.
• airflow.contrib.operators : lieu de résidence des opérateurs de la communauté. Dans ce cas, nous avons besoin de dataproc_operator pour accéder à l'API Cloud Dataproc.
• airflow.utils.trigger_rule : permet d'ajouter des règles de déclencheur à nos opérateurs. Les règles de déclenchement permettent de contrôler de manière précise si un opérateur doit s'exécuter en fonction de l'état de ses parents.

output_file = os.path.join(
    models.Variable.get('gcs_bucket'), 'wordcount',
    datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) + os.sep

Ceci spécifie l'emplacement de notre fichier de sortie. La ligne notable ici est models.Variable.get('gcs_bucket'), qui récupère la valeur de la variable gcs_bucket à partir de la base de données Airflow.

WORDCOUNT_JAR = (
    'file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar'
)

input_file = 'gs://pub/shakespeare/rose.txt'

wordcount_args = ['wordcount', input_file, output_file]

• WORDCOUNT_JAR : emplacement du fichier .jar que nous finirons par exécuter sur le cluster Cloud Dataproc. Il est déjà hébergé sur GCP.
• input_file : emplacement du fichier contenant les données sur lesquelles notre tâche Hadoop finira par effectuer des calculs. Nous importerons ensemble les données à cet emplacement à l'étape 5.
• wordcount_args : arguments que nous allons transmettre dans le fichier JAR.

yesterday = datetime.datetime.combine(
    datetime.datetime.today() - datetime.timedelta(1),
    datetime.datetime.min.time())

Cela nous donnera un objet datetime équivalent représentant minuit la veille. Par exemple, s'il est exécuté à 11 h le 4 mars, l'objet datetime représente alors 00:00 le 3 mars. Cela concerne la manière dont Airflow gère la planification. Pour en savoir plus, cliquez ici.

default_dag_args = {
    'start_date': yesterday,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'project_id': models.Variable.get('gcp_project')
}

La variable default_dag_args sous la forme d'un dictionnaire doit être fournie chaque fois qu'un DAG est créé:

• 'email_on_failure' : indique si des alertes par e-mail doivent être envoyées lorsqu'une tâche a échoué.
• 'email_on_retry' : indique si des alertes par e-mail doivent être envoyées lorsqu'une tâche est relancée.
• 'retries' : indique le nombre de tentatives de nouvelle tentative qu'Airflow doit effectuer en cas d'échec d'un DAG.
• 'retry_delay' : indique la durée pendant laquelle Airflow doit attendre avant de tenter une nouvelle tentative
• 'project_id' : indique au DAG quel ID de projet GCP doit être associé, ce qui sera nécessaire ultérieurement avec l'opérateur Dataproc

with models.DAG(
        'composer_hadoop_tutorial',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

L'utilisation de with models.DAG indique au script d'inclure tout ce qui se trouve en dessous dans le même DAG. Trois arguments sont également transmis:

• Le premier, une chaîne, est le nom à donner au DAG que nous créons. Dans ce cas, nous utilisons composer_hadoop_tutorial.
• schedule_interval : objet datetime.timedelta, que nous avons ici défini sur un jour. Cela signifie que ce DAG tentera de s'exécuter une fois par jour après la 'start_date' définie précédemment dans 'default_dag_args'
• default_args : dictionnaire que nous avons créé précédemment contenant les arguments par défaut du DAG

Créer un cluster Dataproc

Nous allons ensuite créer un dataproc_operator.DataprocClusterCreateOperator qui crée un cluster Cloud Dataproc.

    create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
        task_id='create_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        num_workers=2,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

Cet opérateur contient quelques arguments, tous sauf le premier qui lui sont spécifiques:

• task_id : comme dans BashOperator, il s'agit du nom que nous attribuons à l'opérateur, qui est visible dans l'interface utilisateur Airflow.
• cluster_name : le nom que nous attribuons au cluster Cloud Dataproc. Nous l'avons nommée composer-hadoop-tutorial-cluster-{{ ds_nodash }} (voir la zone d'information pour plus d'informations facultatives).
• num_workers : nombre de nœuds de calcul que nous allouons au cluster Cloud Dataproc
• zone : région géographique dans laquelle nous souhaitons que le cluster soit hébergé, telle qu'elle est enregistrée dans la base de données Airflow. La variable 'gce_zone' définie à l'étape 3 sera lue.
• master_machine_type : type de machine que nous souhaitons allouer au nœud maître Cloud Dataproc
• worker_machine_type : type de machine que nous souhaitons allouer au nœud de calcul Cloud Dataproc

Informations supplémentaires sur la variable cluster_name

La partie {{ ds_nodash }} du paramètre apparaît, car Airflow est compatible avec la modélisation jinja2. Il s'agit d'un paramètre qui est affiché par Airflow au moment de l'exécution chaque fois que l'opérateur démarre. Dans ce cas, {{ ds_nodash }} est remplacé par la date d'exécution du DAG au format AAAAMMJJ. Si vous ne connaissez pas les modèles Jinja2, ce n'est pas grave. Aucune connaissance n'est requise pour cet atelier.

Envoyer une tâche Apache Hadoop

dataproc_operator.DataProcHadoopOperator nous permet d'envoyer une tâche à un cluster Cloud Dataproc.

    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        main_jar=WORDCOUNT_JAR,
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        arguments=wordcount_args)

Nous proposons plusieurs paramètres:

• task_id : nom attribué à cet élément du DAG
• main_jar : emplacement du fichier .jar que vous souhaitez exécuter sur le cluster
• cluster_name : nom du cluster sur lequel exécuter la tâche (comme vous pouvez le constater) : il est identique à celui obtenu dans l'opérateur précédent.
• arguments : arguments transmis dans le fichier JAR, comme vous le feriez si vous exécutiez celui-ci à partir de la ligne de commande

Supprimer le cluster

Le dernier opérateur que nous allons créer est dataproc_operator.DataprocClusterDeleteOperator.

    delete_dataproc_cluster = dataproc_operator.DataprocClusterDeleteOperator(
        task_id='delete_dataproc_cluster',
        cluster_name='composer-hadoop-tutorial-cluster-{{ ds_nodash }}',
        trigger_rule=trigger_rule.TriggerRule.ALL_DONE)

Comme son nom l'indique, cet opérateur supprime un cluster Cloud Dataproc donné. Nous voyons trois arguments ici:

• task_id : comme dans BashOperator, il s'agit du nom que nous attribuons à l'opérateur, qui est visible dans l'interface utilisateur d'Airflow.
• cluster_name : nom que nous attribuons au cluster Cloud Dataproc. Ici, nous l'avons nommé composer-hadoop-tutorial-cluster-{{ ds_nodash }} (voir la zone d'informations après "Créer un cluster Dataproc" pour en savoir plus).
• trigger_rule : au début de cette étape, nous avons brièvement mentionné les règles de déclenchement lors des importations, mais nous en avons une ici en action. Par défaut, un opérateur Airflow ne s'exécute que si tous ses opérateurs en amont ont abouti. La règle de déclencheur ALL_DONE ne nécessite que la fin de tous les opérateurs en amont, qu'ils aient réussi ou non. Ici, cela signifie que même si le job Hadoop échoue, il faut tout de même supprimer le cluster.

  create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster

Enfin, nous voulons que ces opérateurs s'exécutent dans un ordre particulier, et nous pouvons le signaler en utilisant les opérateurs de bitshift Python. Dans ce cas, create_dataproc_cluster s'exécute toujours en premier, suivi de run_dataproc_hadoop, puis de delete_dataproc_cluster.

En regroupant tout, le code se présente comme suit :

# Copyright 2018 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# [START composer_hadoop_tutorial]
"""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 use for result of Hadoop job.
  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.
input_file = 'gs://pub/shakespeare/rose.txt'

wordcount_args = ['wordcount', input_file, 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')
}

# [START composer_hadoop_schedule]
with models.DAG(
        'composer_hadoop_tutorial',
        # Continue to run DAG once per day
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:
    # [END composer_hadoop_schedule]

    # 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,
        zone=models.Variable.get('gce_zone'),
        master_machine_type='n1-standard-1',
        worker_machine_type='n1-standard-1')

    # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster
    # master node.
    run_dataproc_hadoop = dataproc_operator.DataProcHadoopOperator(
        task_id='run_dataproc_hadoop',
        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',
        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)

    # [START composer_hadoop_steps]
    # Define DAG dependencies.
    create_dataproc_cluster >> run_dataproc_hadoop >> delete_dataproc_cluster
    # [END composer_hadoop_steps]

# [END composer_hadoop]

Si vous avez ignoré l'étape 3 parce que votre environnement n'a pas été créé, veillez à y revenir maintenant avant de passer à l'étape 5.

5. Importer des fichiers Airflow dans Cloud Storage

Copiez le DAG dans le dossier /dags

1. Tout d'abord, ouvrez Cloud Shell, sur lequel Cloud SDK est facilement installé.
2. Clonez le dépôt d'exemples Python et accédez au répertoire composer/workflows.

git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git && cd python-docs-samples/composer/workflows

3. Exécutez la commande suivante pour définir le nom de votre dossier de DAG sur une variable d'environnement

DAGS_FOLDER=$(gcloud composer environments describe ${COMPOSER_INSTANCE} \
--location us-central1 --format="value(config.dagGcsPrefix)")

4. Exécutez la commande gsutil suivante pour copier le code de l'atelier dans le répertoire /dags créé.

gsutil cp hadoop_tutorial.py $DAGS_FOLDER

Le résultat doit se présenter comme suit :

Copying file://hadoop_tutorial.py [Content-Type=text/x-python]...
/ [1 files][  4.1 KiB/  4.1 KiB]
Operation completed over 1 objects/4.1 KiB.

Remarque : Ce guide fait désormais référence au workflow sous le nom composer_hadoop_tutorial.

6. Utiliser l'interface utilisateur d'Airflow

Pour accéder à l'interface Web Airflow dans la console GCP : Ouvrez la page Environnements. Dans la colonne Airflow webserver (Serveur Web Airflow) associée à l'environnement, cliquez sur l'icône de nouvelle fenêtre. L'interface utilisateur Web Airflow s'ouvre dans une nouvelle fenêtre du navigateur.

Pour plus d'informations sur l'interface utilisateur d'Airflow, consultez la section Accéder à l'interface Web.

Afficher les variables

Les variables que vous avez définies précédemment sont conservées dans votre environnement. Pour les afficher, sélectionnez Admin > Variables dans la barre de menu de l'interface utilisateur Airflow.

Explorer les exécutions du DAG

Lorsque vous importez votre fichier DAG dans le dossier dags de Cloud Storage, Cloud Composer l'analyse. 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. Pour afficher vos DAG, cliquez sur DAG en haut de la page.

Cliquez sur composer_hadoop_tutorial pour ouvrir la page d'informations du DAG. Cette page comprend une représentation graphique des tâches et des dépendances du workflow.

Maintenant, dans la barre d'outils, cliquez sur Graph View (Vue graphique), puis 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.).

Pour exécuter de nouveau le workflow à partir de la vue graphique, procédez comme suit :

1. Dans la vue graphique de l'interface utilisateur Airflow, cliquez sur le graphique create_dataproc_cluster.
2. Cliquez sur Clear (Effacer) pour réinitialiser les trois tâches, puis sur OK pour confirmer.

Vous pouvez également vérifier l'état et les résultats du workflow composer-hadoop-tutorial en accédant aux pages suivantes de la console GCP :

• Clusters Cloud Dataproc pour surveiller la création et la suppression des clusters. Notez que le cluster créé par le workflow est éphémère: il n'existe que pendant la durée du workflow et il est supprimé lors de la dernière tâche du workflow.
• Jobs Cloud Dataproc pour afficher ou 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.
• Navigateur Cloud Storage pour afficher les résultats du décompte de mots dans le dossier wordcount du bucket Cloud Storage que vous avez créé pour cet atelier de programmation.

7. Nettoyage

Pour éviter que les ressources utilisées dans cet atelier de programmation soient facturées sur votre compte GCP:

1. (Facultatif) Pour enregistrer vos données, téléchargez-les à partir du bucket Cloud Storage de l'environnement Cloud Composer et du bucket de stockage que vous avez créé pour cet atelier de programmation.
2. Supprimez le bucket Cloud Storage que vous avez créé pour cet atelier de programmation.
3. Supprimez le bucket Cloud Storage pour l'environnement.
4. Supprimez l'environnement Cloud Composer. Notez que la suppression de votre environnement n'entraîne pas la suppression du bucket de stockage pour l'environnement.

Vous pouvez également supprimer le projet (facultatif) :

1. Dans la console GCP, accédez à la page Projects (Projets).
2. Dans la liste des projets, sélectionnez celui que vous souhaitez supprimer, puis cliquez sur Supprimer.
3. Dans la zone prévue à cet effet, saisissez l'ID du projet, puis cliquez sur Arrêter pour supprimer le projet.

Attention : La suppression d'un projet aura les effets suivants :

• Tout le contenu du projet est supprimé. Si vous avez utilisé un projet existant pour ce tutoriel et que vous le supprimez, vous supprimerez également tout autre travail effectué dans le projet.
• Les ID de projets personnalisés sont perdus. Lorsque vous avez créé ce projet, vous avez peut-être créé un ID de projet personnalisé que vous souhaitez utiliser à l'avenir. Pour conserver les URL qui utilisent l'ID de projet, telle qu'une URL appspot.com, supprimez les ressources sélectionnées dans le projet au lieu de supprimer l'ensemble du projet.