הרצת DAG של Apache Airflow ב-Managed Airflow (דור 3)

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

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

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

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

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

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

יצירת סביבה

1. נכנסים לדף Create environment במסוף Cloud de Confiance .

   כניסה לדף Create environment

2. בשדה Name, מזינים example-environment.

3. ברשימה הנפתחת Location, בוחרים אזור לסביבת Managed Airflow. במדריך הזה משתמשים באזור us-central1.

4. לאפשרויות אחרות של הגדרת הסביבה, משתמשים בברירות המחדל שסופקו.

5. לוחצים על Create וממתינים עד שהסביבה תיווצר.

6. בסיום, מופיע סימן וי ירוק ליד שם הסביבה.

יצירת קובץ DAG

Airflow DAG הוא אוסף של משימות מאורגנות שרוצים לתזמן ולהריץ. DAGs מוגדרים בקובצי 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 משויכת קטגוריה של Cloud Storage. ‫Airflow ב-Managed Airflow מתזמן רק קובצי DAG שנמצאים בתיקייה /dags בקטגוריה הזו.

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

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

   כניסה לדף Environments

2. ברשימת הסביבות, לוחצים על שם הסביבה, example-environment. הדף Environment details נפתח.

3. לוחצים על Open DAGs folder. הדף Bucket details נפתח.

4. לוחצים על העלאת קבצים ובוחרים את העותק של quickstart.py.

5. כדי להעלות את הקובץ, לוחצים על פתיחה.

הצגת ה-DAG

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

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

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

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

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

   כניסה לדף Environments

3. ברשימת הסביבות, לוחצים על שם הסביבה, example-environment. הדף Environment details נפתח.

4. עוברים לכרטיסייה DAGs.

5. מוודאים שה-DAG‏ composer_quickstart מופיע ברשימת ה-DAGs.

   

הצגת פרטי הרצת DAG

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

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

1. בכרטיסייה DAGs, לוחצים על composer_quickstart. הכרטיסייה Runs של ה-DAG נפתחת.

2. ברשימת הרצות ה-DAG, לוחצים על הרשומה הראשונה.

   

3. פרטי הרצת ה-DAG מוצגים, ומפרטים את המידע על משימות ספציפיות של ה-DAG לדוגמה.

   

4. בקטע Logs for DAG run מופיעים היומנים של כל המשימות בהרצת ה-DAG. אפשר לראות את הפלט של הפקודה echo ביומנים.