הפעלת עומסי עבודה של Managed Service for Apache Spark באמצעות Managed Airflow

Managed Airflow (דור 3) | Managed Airflow (דור 2) | Managed Airflow (דור 1 מדור קודם)

בדף הזה מוסבר איך משתמשים ב-Managed Airflow (דור שני) כדי להריץ עומסי עבודה של Managed Service for Apache Spark ב-Google Cloud.

בדוגמאות שבקטעים הבאים מוסבר איך להשתמש באופרטורים לניהול עומסי עבודה באצווה של Managed Service for Apache Spark. אתם משתמשים באופרטורים האלה ב-DAGs שיוצרים, מוחקים, מציגים ומקבלים עומס עבודה באצווה של Managed Service for Apache Spark:

• יוצרים קובצי DAG לאופרטורים שעובדים עם עומסי עבודה באצווה של Managed Service for Apache Spark:

  • DataprocCreateBatchOperator
  • DataprocDeleteBatchOperator
  • DataprocListBatchesOperator
  • DataprocGetBatchOperator

• ליצור קבוצות DAG שמשתמשות בקונטיינרים בהתאמה אישית וב-Dataproc Metastore.

• מגדירים שרת היסטוריה מתמשכת ל-DAGs האלה.

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

1. מפעילים את Dataproc API:

   המסוף

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

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

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

   להפעלת ה-API

   gcloud

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

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

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

   gcloud services enable dataproc.googleapis.com

2. בוחרים את המיקום של קובץ עומס העבודה באצווה. אתם יכולים להשתמש בכל אחת מהאפשרויות הבאות:

   • יוצרים קטגוריה של Cloud Storage שבה מאחסנים את הקובץ.
   • משתמשים בקטגוריה של הסביבה. מכיוון שאין צורך לסנכרן את הקובץ הזה עם Airflow, אפשר ליצור תיקיית משנה נפרדת מחוץ לתיקיות /dags או /data. לדוגמה, /batches.
   • שימוש בקטגוריה קיימת.

הגדרת קבצים ומשתני Airflow

בקטע הזה נסביר איך להגדיר קבצים ומשתני Airflow בשביל המדריך הזה.

העלאת קובץ של עומס עבודה של Managed Service for Apache Spark ML לקטגוריה

עומס העבודה במדריך הזה מריץ סקריפט pyspark:

1. שומרים סקריפט pyspark בקובץ מקומי בשם spark-job.py. לדוגמה, אתם יכולים להשתמש בסקריפט pyspark לדוגמה.

2. מעלים את הקובץ למיקום שבחרתם בקטע לפני שמתחילים.

הגדרת משתני Airflow

בדוגמאות שבקטעים הבאים נעשה שימוש במשתני Airflow. אתם מגדירים ערכים למשתנים האלה ב-Airflow, ואז קוד ה-DAG יכול לגשת לערכים האלה.

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

מגדירים את משתני Airflow הבאים לשימוש בקוד DAG:

• ‫project_id: מזהה הפרויקט.
• ‫bucket_name: ה-URI של הקטגוריה שבה נמצא קובץ ה-Python הראשי של עומס העבודה (spark-job.py). בחרתם את המיקום הזה בקטע לפני שמתחילים.
• ‫phs_cluster: השם של אשכול Persistent History Server. אתם מגדירים את המשתנה הזה כשאתם יוצרים שרת היסטוריה קבוע.
• ‫image_name: השם והתג של קובץ האימג' של הקונטיינר בהתאמה אישית (image:tag). אתם מגדירים את המשתנה הזה כשאתם משתמשים בקובץ אימג' של קונטיינר בהתאמה אישית עם DataprocCreateBatchOperator.
• ‫metastore_cluster: שם השירות של Dataproc Metastore. אתם מגדירים את המשתנה הזה כשאתם משתמשים בשירות Dataproc Metastore עם DataprocCreateBatchOperator.
• ‫region_name: האזור שבו נמצא שירות Dataproc Metastore. אתם מגדירים את המשתנה הזה כשאתם משתמשים בשירות Dataproc Metastore עם DataprocCreateBatchOperator.

משתמשים במסוף Google Cloud ובממשק המשתמש של Airflow כדי להגדיר כל משתנה של Airflow

1. במסוף Google Cloud , נכנסים לדף Environments.

   כניסה לדף Environments

2. ברשימת הסביבות, לוחצים על הקישור Airflow של הסביבה. ממשק המשתמש של Airflow נפתח.

3. בממשק המשתמש של Airflow, בוחרים באפשרות Admin > Variables.

4. לוחצים על הוספת רשומה חדשה.

5. מציינים את שם המשתנה בשדה Key ומגדירים את הערך שלו בשדה Val.

6. לוחצים על Save.

יצירת שרת היסטוריה קבוע

כדי לראות את קובצי ההיסטוריה של Spark של עומסי העבודה באצווה, אפשר להשתמש בשרת היסטוריה מתמיד (PHS):

1. יצירת שרת היסטוריה קבוע.
2. מוודאים שציינתם את השם של אשכול ה-PHS במשתנה Airflow phs_cluster.

DataprocCreateBatchOperator

ה-DAG הבא מפעיל עומס עבודה באצווה של Managed Service for Apache Spark.

למידע נוסף על הארגומנטים של DataprocCreateBatchOperator, קראו את קוד המקור של האופרטור.

למידע נוסף על מאפיינים שאפשר להעביר בפרמטר batch של DataprocCreateBatchOperator, עיינו בתיאור של המחלקה Batch.

"""
Examples below show how to use operators for managing Dataproc Serverless batch workloads.
 You use these operators in DAGs that create, delete, list, and get a Dataproc Serverless Spark batch workload.
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id is the Google Cloud Project ID to use for the Cloud Dataproc Serverless.
* bucket_name is the URI of a bucket where the main python file of the workload (spark-job.py) is located.
* phs_cluster is the Persistent History Server cluster name.
* image_name is the name and tag of the custom container image (image:tag).
* metastore_cluster is the Dataproc Metastore service name.
* region_name is the region where the Dataproc Metastore service is located.
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataproc import (
    DataprocCreateBatchOperator,
    DataprocDeleteBatchOperator,
    DataprocGetBatchOperator,
    DataprocListBatchesOperator,
)
from airflow.utils.dates import days_ago

PROJECT_ID = "{{ var.value.project_id }}"
REGION = "{{ var.value.region_name}}"
BUCKET = "{{ var.value.bucket_name }}"
PHS_CLUSTER = "{{ var.value.phs_cluster }}"
METASTORE_CLUSTER = "{{var.value.metastore_cluster}}"
DOCKER_IMAGE = "{{var.value.image_name}}"

PYTHON_FILE_LOCATION = "gs://{{var.value.bucket_name }}/spark-job.py"
# for e.g.  "gs//my-bucket/spark-job.py"
# Start a single node Dataproc Cluster for viewing Persistent History of Spark jobs
PHS_CLUSTER_PATH = "projects/{{ var.value.project_id }}/regions/{{ var.value.region_name}}/clusters/{{ var.value.phs_cluster }}"
# for e.g. projects/my-project/regions/my-region/clusters/my-cluster"
SPARK_BIGQUERY_JAR_FILE = "gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar"
# use this for those pyspark jobs that need a spark-bigquery connector
# https://cloud.google.com/dataproc/docs/tutorials/bigquery-connector-spark-example
# Start a Dataproc MetaStore Cluster
METASTORE_SERVICE_LOCATION = "projects/{{var.value.project_id}}/locations/{{var.value.region_name}}/services/{{var.value.metastore_cluster }}"
# for e.g. projects/my-project/locations/my-region/services/my-cluster
CUSTOM_CONTAINER = "us.gcr.io/{{var.value.project_id}}/{{ var.value.image_name}}"
# for e.g. "us.gcr.io/my-project/quickstart-image",

default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "project_id": PROJECT_ID,
    "region": REGION,
}
with models.DAG(
    "dataproc_batch_operators",  # The id you will see in the DAG airflow page
    default_args=default_args,  # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    create_batch = DataprocCreateBatchOperator(
        task_id="batch_create",
        batch={
            "pyspark_batch": {
                "main_python_file_uri": PYTHON_FILE_LOCATION,
                "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
            },
            "environment_config": {
                "peripherals_config": {
                    "spark_history_server_config": {
                        "dataproc_cluster": PHS_CLUSTER_PATH,
                    },
                },
            },
        },
        batch_id="batch-create-phs",
    )
    list_batches = DataprocListBatchesOperator(
        task_id="list-all-batches",
    )

    get_batch = DataprocGetBatchOperator(
        task_id="get_batch",
        batch_id="batch-create-phs",
    )
    delete_batch = DataprocDeleteBatchOperator(
        task_id="delete_batch",
        batch_id="batch-create-phs",
    )
    create_batch >> list_batches >> get_batch >> delete_batch

שימוש בקובץ אימג' של קונטיינר בהתאמה אישית עם DataprocCreateBatchOperator

בדוגמה הבאה מוסבר איך משתמשים בקובץ אימג' של קונטיינר בהתאמה אישית כדי להריץ את עומסי העבודה. אתם יכולים להשתמש בקונטיינר בהתאמה אישית, למשל כדי להוסיף יחסי תלות של Python שלא מסופקים על ידי קובץ האימג' של הקונטיינר שמוגדר כברירת מחדל.

הערה: כדי להפעיל DAGs שמריצים את Managed Service for Apache Spark עם קונטיינרים של Docker בהתאמה אישית, סביבת Managed Airflow צריכה להשתמש ב-Managed Airflow 2.0.6 וב-Airflow 2.2.3, או בגרסאות חדשות יותר.

כדי להשתמש בקובץ אימג' של קונטיינר בהתאמה אישית:

1. יוצרים קובץ אימג' בהתאמה אישית של קונטיינר ומעלים אותו ל-Container Registry.

2. מציינים את התמונה במשתנה Airflow image_name.

3. משתמשים ב-DataprocCreateBatchOperator עם קובץ האימג' בהתאמה אישית:

create_batch_with_custom_container = DataprocCreateBatchOperator(
    task_id="dataproc_custom_container",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
            "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
        },
        "environment_config": {
            "peripherals_config": {
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
        "runtime_config": {
            "container_image": CUSTOM_CONTAINER,
        },
    },
    batch_id="batch-custom-container",
)
get_batch_custom = DataprocGetBatchOperator(
    task_id="get_batch_custom",
    batch_id="batch-custom-container",
)
delete_batch_custom = DataprocDeleteBatchOperator(
    task_id="delete_batch_custom",
    batch_id="batch-custom-container",
)
create_batch_with_custom_container >> get_batch_custom >> delete_batch_custom

שימוש בשירות Dataproc Metastore עם DataprocCreateBatchOperator

כדי להשתמש בשירות Dataproc Metastore מ-DAG:

1. מוודאים ששירות המטא-נתונים כבר פועל.

   במאמר הפעלה והשבתה של Dataproc Metastore מוסבר איך מפעילים שירות מטא-נתונים.

   מידע מפורט על האופרטור של Batch ליצירת ההגדרות זמין במאמר PeripheralsConfig.

2. אחרי ששירות המטא-נתונים פועל, מציינים את השם שלו במשתנה metastore_cluster ואת האזור שלו במשתנה Airflow region_name.

3. שימוש בשירות המטא-נתונים ב-DataprocCreateBatchOperator:

create_batch_with_metastore = DataprocCreateBatchOperator(
    task_id="dataproc_metastore",
    batch={
        "pyspark_batch": {
            "main_python_file_uri": PYTHON_FILE_LOCATION,
            "jar_file_uris": [SPARK_BIGQUERY_JAR_FILE],
        },
        "environment_config": {
            "peripherals_config": {
                "metastore_service": METASTORE_SERVICE_LOCATION,
                "spark_history_server_config": {
                    "dataproc_cluster": PHS_CLUSTER_PATH,
                },
            },
        },
    },
    batch_id="dataproc-metastore",
)
get_batch_metastore = DataprocGetBatchOperator(
    task_id="get_batch_metatstore",
    batch_id="dataproc-metastore",
)
delete_batch_metastore = DataprocDeleteBatchOperator(
    task_id="delete_batch_metastore",
    batch_id="dataproc-metastore",
)

create_batch_with_metastore >> get_batch_metastore >> delete_batch_metastore

DataprocDeleteBatchOperator

אפשר להשתמש ב-DataprocDeleteBatchOperator כדי למחוק אצווה על סמך מזהה האצווה של עומס העבודה.

delete_batch = DataprocDeleteBatchOperator(
    task_id="delete_batch",
    batch_id="batch-create-phs",
)

DataprocListBatchesOperator

‫DataprocListBatchesOperator מציג רשימה של אצוות שקיימות ב-project_id ובאזור נתונים.

list_batches = DataprocListBatchesOperator(
    task_id="list-all-batches",
)

DataprocGetBatchOperator

DataprocGetBatchOperator מאחזר עומס עבודה ספציפי של אצווה.

get_batch = DataprocGetBatchOperator(
    task_id="get_batch",
    batch_id="batch-create-phs",
)

המאמרים הבאים

• סקירה כללית על Managed Service for Apache Spark
• הפעלת DAG באמצעות פונקציות Cloud Run