Hadoop-Wordcount-Job in einem Cloud Dataproc-Cluster ausführen

In dieser Anleitung wird gezeigt, wie Sie mit Managed Service for Apache Airflow einen Apache Airflow-DAG (gerichteter azyklischer Graph) erstellen, der einen Apache Hadoop-Wordcount-Job in einem Managed Service for Apache Spark-Cluster ausführt.

Ziele

1. Greifen Sie auf Ihre Managed Airflow-Umgebung zu und verwenden Sie die Airflow-UI.
2. Airflow-Umgebungsvariablen erstellen und aufrufen
3. DAG mit nachfolgenden Aufgaben erstellen und ausführen:
   1. Erstellt einen Managed Service for Apache Spark-Cluster.
   2. Führt einen Apache Hadoop-Wordcount-Job im Cluster aus.
   3. Gibt die Ergebnisse der Wortzählung in einen Cloud Storage-Bucket aus.
   4. Löscht den Cluster.

Kosten

In diesem Dokument verwenden Sie die folgenden kostenpflichtigen Komponenten von Cloud de Confiance by S3NS:

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

Hinweis

• Prüfen Sie, ob die folgenden APIs in Ihrem Projekt aktiviert sind:

  Console

  Aktivieren Sie die Managed Service for Apache Spark- und Cloud Storage-APIs, falls sie noch nicht aktiviert sind.

  Rollen, die zum Aktivieren von APIs erforderlich sind

  Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Rolle „Inhaber“ (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. Informationen zum Zuweisen von Rollen

  APIs aktivieren

  gcloud

  Aktivieren Sie den Managed Service for Apache Spark und die Cloud Storage APIs, falls noch nicht geschehen:

  Rollen, die zum Aktivieren von APIs erforderlich sind

  Zum Aktivieren von APIs benötigen Sie die Berechtigung serviceusage.services.enable. Wenn Sie das Projekt erstellt haben, haben Sie diese Berechtigung wahrscheinlich bereits über die Rolle „Inhaber“ (roles/owner). Andernfalls können Sie diese Berechtigung über die Rolle „Service Usage-Administrator“ (roles/serviceusage.serviceUsageAdmin) erhalten. Informationen zum Zuweisen von Rollen

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

• Erstellen Sie in Ihrem Projekt einen Cloud Storage-Bucket mit einer beliebigen Speicherklasse und Region, um die Ergebnisse des Hadoop-WordCount-Jobs zu speichern.

• Notieren Sie sich den Pfad des Buckets, den Sie erstellt haben, z. B. gs://example-bucket. Im nächsten Schritt definieren Sie für diesen Pfad eine Airflow-Variable und verwenden sie im Beispiel-DAG.

• Managed Airflow-Umgebung mit Standardparametern erstellen Warten Sie, bis die Erstellung der Umgebung abgeschlossen ist. Wenn der Vorgang abgeschlossen ist, wird links neben dem Umgebungsnamen ein grünes Häkchen angezeigt.

• Notieren Sie sich die Region, in der Sie Ihre Umgebung erstellt haben, z. B. us-central. Im nächsten Schritt definieren Sie für diese Region eine Airflow-Variable und verwenden sie im Beispiel-DAG, um einen Managed Service for Apache Spark-Cluster in derselben Region auszuführen.

Airflow-Variablen festlegen

Airflow-Variablen festlegen, die später im Beispiel-DAG verwendet werden. Sie können beispielsweise Airflow-Variablen in der Airflow-Benutzeroberfläche festlegen.

Beispielworkflow ansehen

Ein Airflow-DAG ist eine Sammlung organisierter Aufgaben, die Sie planen und ausführen möchten. DAGs werden in Standard-Python-Dateien definiert. Der Code in hadoop_tutorial.py ist der Workflowcode.

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

Operatoren

Zur Orchestrierung der drei Aufgaben im Beispielworkflow werden die folgenden drei Airflow-Operatoren für den DAG importiert:

• DataprocClusterCreateOperator: Erstellt einen Managed Service for Apache Spark-Cluster.

• DataProcHadoopOperator: Sendet einen Hadoop-Job zum Zählen von Wörtern und schreibt die Ergebnisse in einen Cloud Storage-Bucket.

• DataprocClusterDeleteOperator: Löscht den Cluster, um laufende Gebühren für Compute Engine zu vermeiden.

Abhängigkeiten

Sie strukturieren die Aufgaben, die Sie ausführen möchten, so, dass die Struktur deren Beziehungen und Abhängigkeiten widerspiegelt. Die Aufgaben in diesem DAG werden sequenziell ausgeführt.

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

Planung

Der Name des DAG ist composer_hadoop_tutorial und der DAG wird einmal täglich ausgeführt. Da start_date, das an default_dag_args übergeben wird, auf yesterday festgelegt ist, plant Managed Airflow den Workflow so, dass er sofort nach dem Hochladen des DAG in den Bucket der Umgebung gestartet wird.

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:

DAG in den Bucket der Umgebung hochladen

In Managed Airflow werden DAGs im Ordner /dags im Bucket Ihrer Umgebung gespeichert.

So laden Sie den DAG hoch:

1. Speichern Sie hadoop_tutorial.py auf Ihrem lokalen Computer.

2. Rufen Sie in der Cloud de Confiance Console die Seite Umgebungen auf.

   Zur Seite Umgebungen

3. Klicken Sie in der Liste der Umgebungen in der Spalte DAGs-Ordner für Ihre Umgebung auf den Link DAGs.

4. Klicken Sie auf Dateien hochladen.

5. Wählen Sie hadoop_tutorial.py auf Ihrem lokalen Computer aus und klicken Sie auf Öffnen.

Managed Airflow fügt den DAG zu Airflow hinzu und plant ihn automatisch. DAG-Änderungen werden innerhalb von 3 bis 5 Minuten wirksam.

DAG-Ausführungen ansehen

Aufgabenstatus ansehen

Wenn Sie Ihre DAG-Datei in den Ordner dags/ in Cloud Storage hochladen, parst Managed Airflow die Datei. Wenn der Vorgang erfolgreich abgeschlossen wurde, wird der Name des Workflows in der DAG-Liste angezeigt und der Workflow wird in die Warteschlange gestellt, um sofort ausgeführt zu werden.

1. Rufen Sie zur Anzeige des Aufgabenstatus die Airflow-Weboberfläche auf und klicken Sie in der Symbolleiste auf DAGs.

   

2. Klicken Sie zum Öffnen der DAG-Detailseite auf composer_hadoop_tutorial. Sie enthält eine grafische Darstellung der Workflowaufgaben und ‑abhängigkeiten.

   

3. Klicken Sie zur Anzeige des Status jeder Aufgabe auf Graph View und bewegen Sie den Mauszeiger auf die Grafik für jede Aufgabe.

   

Workflow wieder in die Warteschlange stellen

So führen Sie den Workflow über die Grafikansicht noch einmal aus:

1. Klicken Sie in der Grafikansicht der Airflow-Benutzeroberfläche auf die Grafik create_dataproc_cluster.
2. Klicken Sie zum Zurücksetzen der drei Aufgaben auf Clear und anschließend auf OK.
3. Klicken Sie in der Grafikansicht noch einmal auf create_dataproc_cluster.
4. Zum nochmaligen Übergeben des Workflows an die Warteschlange klicken Sie auf Run.

Taskergebnisse ansehen

Sie können den Status und die Ergebnisse des composer_hadoop_tutorialWorkflows Cloud de Confiance auch auf den folgenden Konsolenseiten prüfen:

• Managed Service for Apache Spark-Cluster: zum Überwachen der Clustererstellung und ‑löschung. Der vom Workflow erstellte Cluster ist temporär. Er ist nur für die Dauer des Workflows vorhanden und wird im Rahmen der letzten Workflowaufgabe gelöscht.

  Zu Managed Service for Apache Spark-Clustern

• Managed Service for Apache Spark-Jobs zur Anzeige oder zum Monitoring eines Apache Hadoop-Wordcount-Jobs. Klicken Sie auf die Job-ID, um die Ausgabe des Job-Logs aufzurufen.

  Zu Managed Service for Apache Spark-Jobs

• Cloud Storage-Browser: zur Anzeige der Ergebnisse der Wortzählung im Ordner wordcount des Cloud Storage-Bucket, den Sie für diese Anleitung erstellt haben.

  Zum Cloud Storage-Browser

Bereinigen

Löschen Sie die in dieser Anleitung verwendeten Ressourcen:

1. Löschen Sie die Managed Airflow-Umgebung, einschließlich des manuellen Löschens des Buckets der Umgebung.

2. Löschen Sie den Cloud Storage-Bucket, in dem die Ergebnisse des Hadoop-Wordcount-Jobs gespeichert sind.