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

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

במדריך למתחילים הזה נסביר איך ליצור סביבת Managed Service for Apache Airflow ולהריץ DAG של Apache Airflow ב-Managed Airflow (דור 2).

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

  1. נכנסים לחשבון Google Cloud . אם אתם משתמשים חדשים ב- Google Cloud, צרו חשבון כדי שתוכלו להעריך את הביצועים של המוצרים שלנו בתרחישים מהעולם האמיתי. לקוחות חדשים מקבלים בחינם גם קרדיט בשווי 300$ להרצה, לבדיקה ולפריסה של עומסי העבודה.

  2. התקינו את ה-CLI של Google Cloud.

  3. אם אתם משתמשים בספק זהויות חיצוני (IdP), קודם אתם צריכים להיכנס ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.

  4. כדי לאתחל את ה-CLI של gcloud, הריצו את הפקודה הבאה: 

    gcloud init

  5. יוצרים או בוחרים Google Cloud פרויקט.

    תפקידים שנדרשים כדי לבחור או ליצור פרויקט

    • Select a project: כדי לבחור פרויקט לא צריך תפקיד IAM ספציפי – אפשר לבחור כל פרויקט שקיבלתם בו תפקיד.
    • יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד Project Creator (יצירת פרויקטים) (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים

    • יוצרים Google Cloud פרויקט:

      gcloud projects create PROJECT_ID

      מחליפים את PROJECT_ID בשם של פרויקט Google Cloud שיוצרים.

    • בוחרים את הפרויקט שיצרתם: Google Cloud 

      gcloud config set project PROJECT_ID

      מחליפים את PROJECT_ID בשם הפרויקט ב- Google Cloud .

  6. מוודאים שהחיוב מופעל בפרויקט Google Cloud .

  7. התקינו את ה-CLI של Google Cloud.

  8. אם אתם משתמשים בספק זהויות חיצוני (IdP), קודם אתם צריכים להיכנס ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.

  9. כדי לאתחל את ה-CLI של gcloud, הריצו את הפקודה הבאה: 

    gcloud init

  10. יוצרים או בוחרים Google Cloud פרויקט.

    תפקידים שנדרשים כדי לבחור או ליצור פרויקט

    • Select a project: כדי לבחור פרויקט לא צריך תפקיד IAM ספציפי – אפשר לבחור כל פרויקט שקיבלתם בו תפקיד.
    • יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד Project Creator (יצירת פרויקטים) (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים

    • יוצרים Google Cloud פרויקט:

      gcloud projects create PROJECT_ID

      מחליפים את PROJECT_ID בשם של פרויקט Google Cloud שיוצרים.

    • בוחרים את הפרויקט שיצרתם: Google Cloud 

      gcloud config set project PROJECT_ID

      מחליפים את PROJECT_ID בשם הפרויקט ב- Google Cloud .

  11. מוודאים שהחיוב מופעל בפרויקט Google Cloud .

  12. מפעילים את Managed Airflow API:

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

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

    gcloud services enable composer.googleapis.com

  13. כדי לקבל את ההרשאות שדרושות לביצוע ההפעלה המהירה הזו, צריך לבקש מהאדמין להקצות לכם את תפקידי ה-IAM הבאים בפרויקט:

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

    יכול להיות שאפשר לקבל את ההרשאות הנדרשות גם באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש.

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

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

חשבון השירות של הסביבה שלכם הוא לא חשבון משתמש. חשבון שירות הוא סוג מיוחד של חשבון, שאפליקציה או מכונה וירטואלית (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.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.6-airflow-2.11.1 \
    --service-account ENVIRONMENT_SERVICE_ACCOUNT

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

יצירת קובץ DAG

DAG של Airflow הוא אוסף של משימות מאורגנות שרוצים לתזמן ולהריץ. גרפים מכווני מחזור (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 במסוף Google Cloud . ב-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. ‫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: ממשק משתמש של Managed Airflow DAG, ממשק משתמש של 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

הסרת המשאבים

כדי לא לצבור חיובים בחשבון על המשאבים שבהם השתמשתם בדף הזה, אתם צריכים למחוק את הפרויקט יחד עם המשאבים. Google Cloud Google Cloud

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

  1. מוחקים את סביבת Managed Airflow:

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

      מעבר אל Environments

    2. בוחרים באפשרות example-environment ולוחצים על מחיקה.

    3. ממתינים עד שהסביבה תימחק.

  2. מוחקים את קטגוריית הסביבה. מחיקה של סביבת Managed Airflow לא מוחקת את קטגוריית ה-bucket שלה.

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

      עוברים אל Storage > Browser

    2. בוחרים את הקטגוריה של הסביבה ולוחצים על מחיקה. לדוגמה, אפשר לתת לקטגוריה הזו את השם us-central1-example-environ-c1616fe8-bucket.

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