Eseguire un job di conteggio parole Hadoop su un cluster Cloud Dataproc

Questo tutorial mostra come utilizzare Managed Service for Apache Airflow per creare un grafo diretto aciclico (DAG) di Apache Airflow che esegue un job wordcount di Apache Hadoop su un cluster Managed Service for Apache Spark.

Obiettivi

1. Accedi al tuo ambiente Managed Airflow e utilizza la UI di Airflow.
2. Crea e visualizza le variabili di ambiente Airflow.
3. Crea ed esegui un DAG che includa le seguenti attività:
   1. Crea un cluster Managed Service for Apache Spark.
   2. Esegue un job di conteggio parole di Apache Hadoop sul cluster.
   3. Restituisce i risultati del conteggio parole in un bucket Cloud Storage.
   4. Elimina il cluster.

Costi

In questo documento vengono utilizzati i seguenti componenti fatturabili di Cloud de Confiance by S3NS:

• Managed Airflow
• Managed Service for Apache Spark
• Cloud Storage

Prima di iniziare

• Assicurati che nel tuo progetto siano abilitate le seguenti API:

  Console

  Abilita le API Managed Service for Apache Spark e Cloud Storage, se non sono già abilitate.

  Ruoli richiesti per abilitare le API

  Per abilitare le API, devi disporre dell'autorizzazione serviceusage.services.enable. Se hai creato il progetto, probabilmente disponi già di questa autorizzazione tramite il ruolo Proprietario (roles/owner). In caso contrario, puoi ottenere questa autorizzazione tramite il ruolo Amministratore utilizzo dei servizi (roles/serviceusage.serviceUsageAdmin). Scopri come concedere i ruoli.

  Abilita le API

  gcloud

  Abilita le API Managed Service for Apache Spark e Cloud Storage, se non sono già abilitate:

  Ruoli richiesti per abilitare le API

  Per abilitare le API, devi disporre dell'autorizzazione serviceusage.services.enable. Se hai creato il progetto, probabilmente disponi già di questa autorizzazione tramite il ruolo Proprietario (roles/owner). In caso contrario, puoi ottenere questa autorizzazione tramite il ruolo Amministratore utilizzo dei servizi (roles/serviceusage.serviceUsageAdmin). Scopri come concedere i ruoli.

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

• Nel tuo progetto, crea un bucket Cloud Storage di qualsiasi classe di archiviazione e regione per archiviare i risultati del job di conteggio parole di Hadoop.

• Prendi nota del percorso del bucket che hai creato, ad esempio gs://example-bucket. Definirai una variabile Airflow per questo percorso e la utilizzerai nel DAG di esempio più avanti in questo tutorial.

• Crea un ambiente Managed Airflow con parametri predefiniti. Attendi il completamento della creazione dell'ambiente. Al termine, il segno di spunta verde viene visualizzato a sinistra del nome dell'ambiente.

• Prendi nota della regione in cui hai creato l'ambiente, ad esempio us-central. Definirai una variabile Airflow per questa regione e la utilizzerai nel DAG di esempio per eseguire un cluster Managed Service for Apache Spark nella stessa regione.

Imposta le variabili Airflow

Imposta le variabili Airflow da utilizzare in un secondo momento nel DAG di esempio. Ad esempio, puoi impostare le variabili Airflow nella UI di Airflow.

Visualizzare il workflow di esempio

Un DAG di Airflow è una raccolta di attività organizzate che vuoi pianificare ed eseguire. I DAG sono definiti in file Python standard. Il codice mostrato in hadoop_tutorial.py è il codice del flusso di lavoro.

"""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

Operatori

Per orchestrare le tre attività nel workflow di esempio, il DAG importa i seguenti tre operatori Airflow:

• DataprocClusterCreateOperator: crea un cluster Managed Service for Apache Spark.

• DataProcHadoopOperator: invia un job di conteggio parole di Hadoop e scrive i risultati in un bucket Cloud Storage.

• DataprocClusterDeleteOperator: elimina il cluster per evitare addebiti continui per Compute Engine.

Dipendenze

Organizzi le attività da eseguire in modo da riflettere le rispettive relazioni e dipendenze. Le attività in questo DAG vengono eseguite in sequenza.

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

Programmazione

Il nome del DAG è composer_hadoop_tutorial e il DAG viene eseguito una volta al giorno. Poiché start_date passato a default_dag_args è impostato su yesterday, Managed Airflow pianifica l'avvio del workflow subito dopo il caricamento del DAG nel bucket dell'ambiente.

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:

Carica il DAG nel bucket dell'ambiente

Managed Airflow archivia i DAG nella cartella /dags del bucket dell'ambiente.

Per caricare il DAG:

1. Salva hadoop_tutorial.py sulla tua macchina locale.

2. Nella console Cloud de Confiance , vai alla pagina Ambienti.

   Vai ad Ambienti

3. Nell'elenco degli ambienti, nella colonna Cartella DAG per il tuo ambiente, fai clic sul link DAG.

4. Fai clic su Carica file.

5. Seleziona hadoop_tutorial.py sulla macchina locale e fai clic su Apri.

Managed Airflow aggiunge il DAG ad Airflow e lo pianifica automaticamente. Le modifiche ai DAG vengono applicate entro 3-5 minuti.

Esplora le esecuzioni del DAG

Visualizzare lo stato dell'attività

Quando carichi il file DAG nella cartella dags/ di Cloud Storage, Managed Airflow analizza il file. Al termine, il nome del workflow viene visualizzato nell'elenco di DAG e il workflow viene messo in coda per essere eseguito immediatamente.

1. Per visualizzare lo stato dell'attività, vai all'interfaccia web di Airflow e fai clic su DAG nella barra degli strumenti.

   

2. Per aprire la pagina dei dettagli del DAG, fai clic su composer_hadoop_tutorial. Questa pagina include una rappresentazione grafica delle attività e delle dipendenze del workflow.

   

3. Per visualizzare lo stato di ogni attività, fai clic su Visualizzazione a grafo e poi passa il mouse sopra il grafico di ogni attività.

   

Mettere di nuovo in coda il workflow

Per eseguire di nuovo il workflow dalla visualizzazione Grafico:

1. Nella visualizzazione a grafo della UI di Airflow, fai clic sul grafico create_dataproc_cluster.
2. Per reimpostare le tre attività, fai clic su Cancella e poi su Ok per confermare.
3. Fai di nuovo clic su create_dataproc_cluster nella visualizzazione Grafico.
4. Per mettere di nuovo in coda il flusso di lavoro, fai clic su Esegui.

Visualizzare i risultati dell'attività

Puoi anche controllare lo stato e i risultati del workflow composer_hadoop_tutorial andando alle seguenti pagine della console Cloud de Confiance :

• Cluster Managed Service for Apache Spark: per monitorare la creazione e l'eliminazione dei cluster. Tieni presente che il cluster creato dal workflow è temporaneo: esiste solo per la durata del workflow e viene eliminato dall'ultima attività del workflow.

  Vai ai cluster Managed Service for Apache Spark

• Managed Service for Apache Spark Jobs: per visualizzare o monitorare il job di conteggio parole di Apache Hadoop. Fai clic sull'ID job per visualizzarne l'output del log.

  Vai ai job Managed Service for Apache Spark

• Browser Cloud Storage: per visualizzare i risultati del conteggio delle parole nella cartella wordcount del bucket Cloud Storage che hai creato per questo tutorial.

  Vai a Browser Cloud Storage.

Esegui la pulizia

Elimina le risorse utilizzate in questo tutorial:

1. Elimina l'ambiente Managed Airflow, incluso l'eliminazione manuale del bucket dell'ambiente.

2. Elimina il bucket Cloud Storage che memorizza i risultati del job di conteggio parole di Hadoop.