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

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Ce tutoriel explique comment utiliser Cloud Composer pour créer un Graphe orienté acyclique (DAG) pour Apache Airflow qui exécute un job de décompte de mots Apache Hadoop sur Dataproc cluster.

Objectifs

1. Accédez à votre environnement Cloud Composer et utilisez l'interface utilisateur Airflow.
2. Créer et afficher des variables d'environnement Airflow
3. Créer et exécuter un DAG comprenant les tâches suivantes :
   1. Crée un cluster Dataproc.
   2. Exécution d'une tâche de décompte de mots Apache Hadoop sur le cluster
   3. Génère les résultats du nombre de mots dans un Cloud Storage bucket.
   4. Supprime le cluster.

Coûts

Dans ce document, vous utilisez les composants facturables suivants de Google Cloud :

• Cloud Composer
• Dataproc
• Cloud Storage

Obtenez une estimation des coûts en fonction de votre utilisation prévue à l'aide du simulateur de coût. Les nouveaux utilisateurs de Google Cloud peuvent bénéficier d'un essai gratuit.

Avant de commencer

• Assurez-vous que les API suivantes sont activées dans votre projet:

  Console

  Enable the Dataproc, Cloud Storage APIs.

  Enable the APIs

  gcloud

  Enable the Dataproc, Cloud Storage APIs:

  gcloud services enable dataproc.googleapis.com storage-component.googleapis.com

• Dans votre projet, créez un bucket Cloud Storage de n'importe quelle classe de stockage et de région pour stocker les résultats de la tâche Hadoop de décompte de mots.

  Remarque : Pour respecter les exigences de résidence de données ou pour bénéficier d'avantages en termes de performances, créez le bucket de stockage dans la région dans laquelle vous prévoyez de créer votre environnement.

• Notez le chemin du bucket que vous avez créé, par exemple gs://example-bucket. Vous allez définir une variable Airflow pour ce chemin et l'utiliser dans l'exemple de DAG plus tard dans ce tutoriel.

• Créez un environnement Cloud Composer avec les paramètres par défaut. Attendez la fin de la création de l'environnement. Une fois l'opération terminée, une coche verte s'affiche à gauche du nom de l'environnement.

• Notez la région dans laquelle vous avez créé votre environnement, par exemple us-central. Vous allez définir une variable Airflow pour cette région et l'utiliser dans l'exemple de DAG pour exécuter un cluster Dataproc dans la même région.

Définir les variables Airflow

Définissez les variables Airflow à utiliser ultérieurement dans l'exemple de DAG. Par exemple : Vous pouvez définir des variables Airflow dans l'interface utilisateur Airflow.

Variable Airflow	Valeur
gcp_project	ID de projet du projet que vous utilisez pour ce tutoriel, par exemple example-project.
gcs_bucket	Le bucket URI Cloud Storage que vous avez créé pour ce tutoriel comme gs://example-bucket.
gce_region	Région dans laquelle vous avez créé votre environnement, par exemple us-central1. Il s'agit de la région dans laquelle votre cluster Dataproc sera créé.

Afficher l'exemple de workflow

Un DAG Airflow est un ensemble de tâches organisées que vous souhaitez programmer et exécuter. Les DAG sont définis dans des fichiers Python standards. Le code présenté dans hadoop_tutorial.py est le code du workflow.

Airflow 2

"""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/docs/apache-airflow/stable/concepts/variables.html
* gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster.
* gce_region - Google Compute Engine region 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.providers.google.cloud.operators import dataproc
from airflow.utils import trigger_rule

# Output file for Cloud Dataproc job.
# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
output_file = (
    os.path.join(
        "{{ var.value.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]

HADOOP_JOB = {
    "reference": {"project_id": "{{ var.value.gcp_project }}"},
    "placement": {"cluster_name": "composer-hadoop-tutorial-cluster-{{ ds_nodash }}"},
    "hadoop_job": {
        "main_jar_file_uri": WORDCOUNT_JAR,
        "args": wordcount_args,
    },
}

CLUSTER_CONFIG = {
    "master_config": {"num_instances": 1, "machine_type_uri": "n1-standard-2"},
    "worker_config": {"num_instances": 2, "machine_type_uri": "n1-standard-2"},
}

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": "{{ var.value.gcp_project }}",
    "region": "{{ var.value.gce_region }}",
}


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:

    # Create a Cloud Dataproc cluster.
    create_dataproc_cluster = dataproc.DataprocCreateClusterOperator(
        task_id="create_dataproc_cluster",
        # Give the cluster a unique name by appending the date scheduled.
        # See https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html
        cluster_name="composer-hadoop-tutorial-cluster-{{ ds_nodash }}",
        cluster_config=CLUSTER_CONFIG,
        region="{{ var.value.gce_region }}",
    )

    # Run the Hadoop wordcount example installed on the Cloud Dataproc cluster
    # master node.
    run_dataproc_hadoop = dataproc.DataprocSubmitJobOperator(
        task_id="run_dataproc_hadoop", job=HADOOP_JOB
    )

    # Delete Cloud Dataproc cluster.
    delete_dataproc_cluster = dataproc.DataprocDeleteClusterOperator(
        task_id="delete_dataproc_cluster",
        cluster_name="composer-hadoop-tutorial-cluster-{{ ds_nodash }}",
        region="{{ var.value.gce_region }}",
        # 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

Airflow 1

"""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/docs/apache-airflow/stable/concepts/variables.html
* gcp_project - Google Cloud Project to use for the Cloud Dataproc cluster.
* gce_region - Google Compute Engine region 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.
# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
output_file = (
    os.path.join(
        "{{ var.value.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": "{{ var.value.gcp_project }}",
}

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:

    # 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/docs/apache-airflow/stable/macros-ref.html
        cluster_name="composer-hadoop-tutorial-cluster-{{ ds_nodash }}",
        num_workers=2,
        region="{{ var.value.gce_region }}",
        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",
        main_jar=WORDCOUNT_JAR,
        region="{{ var.value.gce_region }}",
        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 }}",
        region="{{ var.value.gce_region }}",
        # 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

Opérateurs

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

• DataprocClusterCreateOperator : crée un cluster Dataproc.

• DataProcHadoopOperator : envoie une tâche Hadoop de décompte de mots et écrit les résultats dans un bucket Cloud Storage.

• DataprocClusterDeleteOperator : supprime le cluster pour éviter que des frais Compute Engine ne continuent d'être facturés.

Dépendances

Vous organisez les tâches que vous souhaitez exécuter de manière à refléter les relations et les dépendances. Les tâches de ce DAG sont exécutées de manière séquentielle.

Airflow 2

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

Airflow 1

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

Planification

Le nom du DAG est composer_hadoop_tutorial, et le DAG s'exécute une fois par jour. Comme la valeur start_date transmise à default_dag_args est définie sur yesterday, Cloud Composer programme le workflow de façon qu'il commence immédiatement après l'importation du DAG dans le bucket de l'environnement.

Airflow 2

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:

Airflow 1

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:

Importer le DAG dans le bucket de l'environnement

Cloud Composer stocke les DAG dans le dossier /dags du bucket de votre environnement.

Pour importer le DAG :

1. Sur votre ordinateur local, enregistrez hadoop_tutorial.py.

2. Dans la console Google Cloud, accédez à la page Environnements.

   Accéder à la page Environnements

3. Dans la liste des environnements, dans la colonne DAGs folder (Dossier des DAG) de votre cliquez sur le lien DAG.

4. Cliquez sur Importer des fichiers.

5. Sélectionnez hadoop_tutorial.py sur votre machine locale, puis cliquez sur Ouvrir.

Cloud Composer ajoute le DAG à Airflow et le planifie automatiquement. Les modifications sont appliquées au DAG après 3 à 5 minutes.

Explorer les exécutions du DAG

Afficher l'état des tâches

Lorsque vous importez votre fichier DAG dans le dossier dags/ de Cloud Storage, Cloud Composer analyse le fichier. Une fois l'opération terminée, le nom du workflow apparaît dans la liste des DAG, et il est mis en file d'attente en vue de son exécution immédiatement.

1. Pour connaître l'état des tâches, accédez à l'interface Web Airflow, puis cliquez sur DAGs (DAG) dans la barre d'outils.

   

2. Pour ouvrir la page de détails des DAG, cliquez sur composer_hadoop_tutorial. Ce comprend une représentation graphique des tâches les dépendances.

   

3. Pour connaître l'état de chaque tâche, cliquez sur Graph View (Vue graphique), puis sur le graphique de chaque tâche.

   

Mettre à nouveau le workflow en file d'attente

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. Pour réinitialiser les trois tâches, cliquez sur Clear (Effacer), puis sur OK pour confirmer.
3. Cliquez de nouveau sur create_dataproc_cluster dans la vue graphique.
4. Pour remettre le workflow en file d'attente, cliquez sur Run (Exécuter).

Afficher les résultats des tâches

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 Google Cloud :

• Clusters Dataproc: pour surveiller la création des clusters et de suppression. Notez que le cluster créé par le workflow est éphémère : n'existe que pendant la durée du workflow et est supprimé la dernière tâche du workflow.

  Accéder aux clusters Dataproc

• Tâches Dataproc : pour afficher ou surveiller la tâche de décompte de mots Apache Hadoop. Cliquez sur l'ID de job pour afficher la sortie du journal associée au job.

  Accéder à "Tâches Dataproc"

• Navigateur Cloud Storage: pour afficher les résultats du décompte de mots dans le dossier wordcount dans le bucket Cloud Storage que vous avez créé pour ce tutoriel.

  Accéder au navigateur Cloud Storage

Nettoyage

Supprimez les ressources utilisées dans ce tutoriel:

1. Supprimez l'environnement Cloud Composer, y compris supprimer manuellement le bucket de l'environnement.

2. Supprimez le bucket Cloud Storage qui stocke les résultats de la tâche de décompte de mots Hadoop.