הפעלת DAG של Apache Airflow ב-Managed Airflow (דור 2) (Google Cloud CLI)

יצירת חשבון שירות של סביבה

כשיוצרים סביבה, מציינים חשבון שירות. חשבון השירות הזה נקרא חשבון השירות של הסביבה. בסביבה שלכם נעשה שימוש בחשבון השירות הזה כדי לבצע את רוב הפעולות.

חשבון השירות של הסביבה שלכם הוא לא חשבון משתמש. חשבון שירות הוא סוג מיוחד של חשבון, שאפליקציה או מכונה וירטואלית (VM) משתמשות בו, ולא אדם.

כדי ליצור חשבון שירות לסביבה שלכם:

1. יוצרים חשבון שירות חדש, כמו שמתואר במסמכי התיעוד של ניהול הזהויות והרשאות הגישה.

2. מקצים לו תפקיד, כמו שמתואר בחומרי העזר של ניהול הזהויות והרשאות הגישה (IAM). התפקיד הנדרש הוא Composer Worker (composer.worker).

יצירת סביבה

אם זו הסביבה הראשונה בפרויקט, צריך להוסיף את חשבון הסוכן של Managed Airflow Service בתור חשבון משתמש חדש בחשבון השירות של הסביבה, ולהקצות לו את התפקיד roles/composer.ServiceAgentV2Ext.

כברירת מחדל, הסביבה שלכם משתמשת בחשבון השירות המוגדר כברירת מחדל ב-Compute Engine, ובדוגמה הבאה מוסבר איך להוסיף לו את ההרשאה הזו.

# Get current project's project number
PROJECT_NUMBER=$(gcloud projects list \
  --filter="$(gcloud config get-value project)" \
  --format="value(PROJECT_NUMBER)" \
  --limit=1)

# Add the Cloud Composer v2 API Service Agent Extension role
gcloud iam service-accounts add-iam-policy-binding \
    ENVIRONMENT_SERVICE_ACCOUNT \
    --member serviceAccount:service-$PROJECT_NUMBER@cloudcomposer-accounts.s3ns-system.iam.gserviceaccount.com \
    --role roles/composer.ServiceAgentV2Ext

מחליפים את ENVIRONMENT_SERVICE_ACCOUNT בחשבון השירות של הסביבה שיצרתם קודם.

יוצרים סביבה חדשה בשם example-environment באזור us-central1, עם הגרסה האחרונה של Managed Airflow (דור 2).

gcloud composer environments create example-environment \
    --location us-central1 \
    --image-version composer-2.17.15-airflow-2.11.1 \
    --service-account ENVIRONMENT_SERVICE_ACCOUNT

מחליפים את ENVIRONMENT_SERVICE_ACCOUNT בחשבון השירות של הסביבה שיצרתם קודם.

יצירת קובץ DAG

Airflow DAG הוא אוסף של משימות מאורגנות שרוצים לתזמן ולהריץ. ‫DAG מוגדרים בקובצי Python רגילים.

במדריך הזה נעשה שימוש בדוגמה ל-DAG של Airflow שמוגדר בקובץ quickstart.py. קוד Python בקובץ הזה מבצע את הפעולות הבאות:

1. יוצר DAG, ‏ composer_sample_dag. ה-DAG הזה מופעל כל יום.
2. מבצע משימה אחת, print_dag_run_conf. המשימה מדפיסה את ההגדרה של הפעלת ה-DAG באמצעות אופרטור bash.

שומרים עותק של קובץ quickstart.py במחשב המקומי:

import datetime

from airflow import models
from airflow.operators import bash

# 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
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    "owner": "Composer Example",
    "depends_on_past": False,
    "email": [""],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "start_date": YESTERDAY,
}

with models.DAG(
    "composer_quickstart",
    catchup=False,
    default_args=default_args,
    schedule_interval=datetime.timedelta(days=1),
) as dag:
    # Print the dag_run id from the Airflow logs
    print_dag_run_conf = bash.BashOperator(
        task_id="print_dag_run_conf", bash_command="echo {{ dag_run.id }}"
    )

העלאת קובץ ה-DAG לקטגוריה של הסביבה

לכל סביבת Managed Airflow משויך מאגר (bucket) של Cloud Storage. ב-Airflow ב-Managed Airflow, מערכות מתזמנות רק DAG שנמצאים בתיקייה /dags בדלי הזה.

כדי לתזמן את ה-DAG, מעלים את quickstart.py מהמחשב המקומי לתיקייה /dags בסביבה:

כדי להעלות את quickstart.py באמצעות Google Cloud CLI, מריצים את הפקודה הבאה בתיקייה שבה נמצא הקובץ quickstart.py:

gcloud composer environments storage dags import \
--environment example-environment --location us-central1 \
--source quickstart.py

הצגת ה-DAG

אחרי שמעלים את קובץ ה-DAG, Airflow מבצע את הפעולות הבאות:

1. מנתח את קובץ ה-DAG שהעליתם. יכול להיות שיחלפו כמה דקות עד ש-DAG יהיה זמין ל-Airflow.
2. ה-DAG יתווסף לרשימת ה-DAG הזמינים.
3. מריץ את ה-DAG בהתאם ללוח הזמנים שציינתם בקובץ ה-DAG.

כדי לוודא ש-DAG עובד ללא שגיאות וזמין ב-Airflow, אפשר להציג אותו בממשק המשתמש של DAG. ממשק המשתמש של DAG הוא ממשק Managed Airflow לצפייה בפרטי DAG במסוף Cloud de Confiance . ב-Managed Airflow יש גם גישה לממשק המשתמש של Airflow, שהוא ממשק אינטרנט מקורי של Airflow.

1. מחכים כחמש דקות כדי לתת ל-Airflow זמן לעבד את קובץ ה-DAG שהעליתם קודם, ולהשלים את ההרצה הראשונה של ה-DAG (מוסבר בהמשך).

2. מריצים את הפקודה הבאה ב-CLI של Google Cloud. הפקודה הזו מריצה את dags list הפקודה Airflow CLI שמציגה רשימה של DAG בסביבה שלכם.

   gcloud composer environments run example-environment \
   --location us-central1 \
   dags list
   

3. מוודאים ש-composer_quickstart DAG מופיע בפלט של הפקודה.

   פלט לדוגמה:

   ‫

   Executing the command: [ airflow dags list ]...
   Command has been started. execution_id=d49074c7-bbeb-4ee7-9b26-23124a5bafcb
   Use ctrl-c to interrupt the command
   dag_id              | filepath              | owner            | paused
   ====================+=======================+==================+=======
   airflow_monitoring  | airflow_monitoring.py | airflow          | False
   composer_quickstart | dag-quickstart-af2.py | Composer Example | False
   

צפייה בפרטי ההרצה של DAG

כל הפעלה של DAG נקראת DAG run. ‫Airflow מריץ באופן מיידי את ה-DAG לדוגמה כי תאריך ההתחלה בקובץ ה-DAG מוגדר ליום האתמול. כך, Airflow מתעדכן בהתאם לתזמון של ה-DAG שצוין.

ה-DAG לדוגמה מכיל משימה אחת, print_dag_run_conf, שמריצה את הפקודה echo במסוף. הפקודה הזו מחזירה מידע על ה-DAG (המזהה המספרי של הרצת ה-DAG).

מריצים את הפקודה הבאה ב-CLI של Google Cloud. הפקודה הזו מציגה רשימה של הפעלות DAG עבור ה-DAG‏ composer_quickstart:

gcloud composer environments run example-environment \
--location us-central1 \
dags list-runs -- --dag-id composer_quickstart

פלט לדוגמה:

dag_id              | run_id                                      | state   | execution_date                   | start_date                       | end_date
====================+=============================================+=========+==================================+==================================+=================================
composer_quickstart | scheduled__2024-02-17T15:38:38.969307+00:00 | success | 2024-02-17T15:38:38.969307+00:00 | 2024-02-18T15:38:39.526707+00:00 | 2024-02-18T15:38:42.020661+00:00

ב-Airflow CLI אין פקודה להצגת יומני משימות. אפשר להשתמש בשיטות אחרות כדי לצפות ביומני משימות של Airflow: ממשק משתמש של DAG של Managed Airflow, ממשק משתמש של Airflow או Cloud Logging. במדריך הזה מוצגת דרך לשאילתות ב-Cloud Logging לגבי יומנים מהרצה ספציפית של DAG.

מריצים את הפקודה הבאה ב-CLI של Google Cloud. הפקודה הזו קוראת יומנים מ-Cloud Logging להרצה ספציפית של DAG של composer_quickstart. הארגומנט --format מעצב את הפלט כך שיוצג רק הטקסט של הודעת היומן.

gcloud logging read \
--format="value(textPayload)" \
--order=asc \
"resource.type=cloud_composer_environment \
resource.labels.location=us-central1 \
resource.labels.environment_name=example-environment \
labels.workflow=composer_quickstart \
(labels.\"execution-date\"=\"RUN_ID\")"

מחליפים את:

• ‫RUN_ID עם הערך run_id מהפלט של הפקודה tasks states-for-dag-run שהפעלתם קודם. לדוגמה, 2024-02-17T15:38:38.969307+00:00.

פלט לדוגמה:

...

Starting attempt 1 of 2
Executing <Task(BashOperator): print_dag_run_conf> on 2024-02-17
15:38:38.969307+00:00
Started process 22544 to run task

...

Running command: ['/usr/bin/bash', '-c', 'echo 115746']
Output:
115746

...

Command exited with return code 0
Marking task as SUCCESS. dag_id=composer_quickstart,
task_id=print_dag_run_conf, execution_date=20240217T153838,
start_date=20240218T153841, end_date=20240218T153841
Task exited with return code 0
0 downstream tasks scheduled from follow-on schedule check