הרצת משימה של ספירת מילים ב-Hadoop באשכול של Cloud Dataproc

במדריך הזה נסביר איך להשתמש ב-Managed Service for Apache Airflow כדי ליצור DAG ‏ (גרף אציקלי מכוון) של Apache Airflow שמריץ משימת ספירת מילים של Apache Hadoop באשכול Managed Service for Apache Spark.

מטרות

1. ניגשים לסביבת Managed Airflow ומשתמשים בממשק המשתמש של Airflow.
2. יצירה של משתני סביבה ב-Airflow וצפייה בהם.
3. יצירה והרצה של DAG שכולל את המשימות הבאות:
   1. יוצר אשכול של Managed Service for Apache Spark.
   2. מריץ משימה של ספירת מילים ב-Apache Hadoop באשכול.
   3. התוצאות של ספירת המילים מועברות לקטגוריה של Cloud Storage.
   4. מחיקת האשכול.

עלויות

במסמך הזה משתמשים ברכיבים הבאים של Cloud de Confiance by S3NS, והשימוש בהם כרוך בתשלום:

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

לפני שמתחילים

• מוודאים שממשקי ה-API הבאים מופעלים בפרויקט:

  המסוף

  מפעילים את Managed Service for Apache Spark ואת Cloud Storage APIs, אם הם עדיין לא מופעלים.

  תפקידים שנדרשים להפעלת ממשקי API

  כדי להפעיל ממשקי API, נדרשת ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין בממשק Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים

  הפעלת ממשקי ה-API

  gcloud

  מפעילים את Managed Service for Apache Spark ואת Cloud Storage APIs, אם הם עדיין לא מופעלים:

  תפקידים שנדרשים להפעלת ממשקי API

  כדי להפעיל ממשקי API, נדרשת ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין בממשק Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים

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

• בפרויקט, יוצרים קטגוריה של Cloud Storage עם סוג אחסון ואזור כלשהם כדי לאחסן את התוצאות של משימת ספירת המילים של Hadoop.

• שימו לב לנתיב של הקטגוריה שיצרתם, לדוגמה gs://example-bucket. בהמשך המדריך הזה תגדירו משתנה Airflow לנתיב הזה ותשתמשו במשתנה ב-DAG לדוגמה.

• יצירת סביבת Managed Airflow עם פרמטרים שמוגדרים כברירת מחדל. מחכים עד ליצירת הסביבה. בסיום, סימן וי ירוק מופיע מימין לשם הסביבה.

• שימו לב לאזור שבו יצרתם את הסביבה, למשל us-central. תגדירו משתנה Airflow לאזור הזה ותשתמשו בו ב-DAG לדוגמה כדי להפעיל אשכול Managed Service for Apache Spark באותו אזור.

הגדרת משתנים ב-Airflow

מגדירים את המשתנים של Airflow כדי להשתמש בהם בהמשך בדוגמה של DAG. לדוגמה, אפשר להגדיר משתני Airflow בממשק המשתמש של Airflow.

דוגמה לזרימת עבודה

‫DAG ב-Airflow הוא אוסף של משימות מאורגנות שרוצים לתזמן ולהריץ. גרפים מכווני מחזור (DAG) מוגדרים בקובצי Python רגילים. הקוד שמוצג ב-hadoop_tutorial.py הוא קוד תהליך העבודה.

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

אופרטורים

כדי לתזמן את שלוש המשימות בתהליך העבודה לדוגמה, ה-DAG מייבא את שלושת האופרטורים הבאים של Airflow:

• ‫DataprocClusterCreateOperator: יצירת אשכול של Managed Service for Apache Spark.

• ‫DataProcHadoopOperator: שולח משימת ספירת מילים של Hadoop וכותב את התוצאות לקטגוריה של Cloud Storage.

• ‫DataprocClusterDeleteOperator: מחיקת האשכול כדי להימנע מחיובים שוטפים ב-Compute Engine.

פניות קשורות

אתם מארגנים את המשימות שאתם רוצים להריץ באופן שמשקף את הקשרים ויחסי התלות ביניהן. המשימות ב-DAG הזה רצות ברצף.

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

קביעת מועד

השם של ה-DAG הוא composer_hadoop_tutorial, והוא מופעל פעם ביום. כי הערך של start_date שמועבר אל default_dag_args מוגדר כ-yesterday. מערכת Managed Airflow מתזמנת את תהליך העבודה כך שיתחיל מיד אחרי שה-DAG מועלה אל הדלי של הסביבה.

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 לקטגוריה של הסביבה

‫Managed Airflow מאחסן DAGs בתיקייה /dags בדלי של הסביבה.

כדי להעלות את ה-DAG:

1. במחשב המקומי, שומרים את הקובץ hadoop_tutorial.py.

2. נכנסים לדף Environments במסוף Cloud de Confiance .

   מעבר אל Environments

3. ברשימת הסביבות, בעמודה DAGs folder של הסביבה, לוחצים על הקישור DAGs.

4. לוחצים על העלאת קבצים.

5. בוחרים באפשרות hadoop_tutorial.py במחשב המקומי ולוחצים על פתיחה.

‫Managed Airflow מוסיף את ה-DAG ל-Airflow ומתזמן את ה-DAG באופן אוטומטי. השינויים ב-DAG מתעדכנים תוך 3-5 דקות.

עיון בהרצות של DAG

צפייה בסטטוס המשימה

כשמעלים את קובץ ה-DAG לתיקייה dags/ ב-Cloud Storage,‏ Managed Airflow מנתח את הקובץ. אם הפעולה הושלמה בהצלחה, השם של תהליך העבודה מופיע ברשימת ה-DAG, ותהליך העבודה מתווסף לתור להפעלה מיידית.

1. כדי לראות את סטטוס המשימה, עוברים לממשק האינטרנט של Airflow ולוחצים על DAGs בסרגל הכלים.

   

2. כדי לפתוח את דף הפרטים של ה-DAG, לוחצים על composer_hadoop_tutorial. בדף הזה יש ייצוג גרפי של משימות ויחסי תלות בתהליך העבודה.

   

3. כדי לראות את הסטטוס של כל משימה, לוחצים על תצוגת תרשים ומעבירים את העכבר מעל הגרפיקה של כל משימה.

   

הוספה חוזרת של תהליך העבודה לתור

כדי להריץ שוב את תהליך העבודה מתצוגת התרשים:

1. בתצוגת הגרף של ממשק המשתמש של Airflow, לוחצים על הגרפיקה create_dataproc_cluster.
2. כדי לאפס את שלוש המשימות, לוחצים על ניקוי ואז על אישור כדי לאשר.
3. לוחצים שוב על create_dataproc_cluster בתצוגת התרשים.
4. כדי להוסיף שוב את תהליך העבודה לתור, לוחצים על הפעלה.

צפייה בתוצאות של משימות

אפשר גם לבדוק את הסטטוס והתוצאות של composer_hadoop_tutorial זרימת העבודה בדפים הבאים במסוף: Cloud de Confiance

• ‫Managed Service for Apache Spark Clusters: כדי לעקוב אחרי יצירה ומחיקה של אשכולות. הערה: האשכול שנוצר על ידי תהליך העבודה הוא זמני: הוא קיים רק למשך תהליך העבודה ונמחק כחלק מהמשימה האחרונה בתהליך העבודה.

  מעבר אל Managed Service for Apache Spark Clusters

• ‫Managed Service for Apache Spark Jobs: כדי להציג או לנטר את משימת ספירת המילים של Apache Hadoop. לוחצים על מזהה משימה כדי לראות את הפלט של יומן העבודה.

  מעבר אל Managed Service for Apache Spark Jobs

• ‫Cloud Storage Browser: כדי לראות את התוצאות של ספירת המילים בתיקייה wordcount בקטגוריה של Cloud Storage שיצרתם לצורך המדריך הזה.

  כניסה לדף Cloud Storage Browser

הסרת המשאבים

מוחקים את המשאבים שבהם השתמשתם במדריך הזה:

1. מחיקת סביבת Airflow מנוהלת, כולל מחיקה ידנית של קטגוריית הסביבה.

2. מוחקים את קטגוריית Cloud Storage שבה מאוחסנות התוצאות של משימת ספירת המילים ב-Hadoop.