תזמון והפעלה של DAG ב-Airflow

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

בדף הזה נסביר איך עובד התזמון והפעלת DAG ב-Airflow, איך מגדירים תזמון ל-DAG ואיך מפעילים DAG באופן ידני או משהים אותו.

מידע על Airflow DAGs ב-Managed Airflow

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

מידע נוסף על מושגי הליבה של Airflow, כמו Airflow DAGs,‏ DAG runs, ‏ tasks או operators, זמין בדף Core Concepts במסמכי התיעוד של Airflow.

מידע על תזמון DAG ב-Airflow

מנגנון התזמון של Airflow מבוסס על המושגים הבאים:

תאריך לוגי

מייצג תאריך שבו מופעלת הרצה מסוימת של DAG.

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

בגרסאות Airflow קודמות לגרסה 2.2, התאריך הזה נקרא תאריך ההפעלה.

תאריך ההפעלה

מייצג תאריך שבו מופעל ריצת DAG מסוימת.

לדוגמה, אם DAG מתוזמן להפעלה כל יום בשעה 12:00, ההפעלה בפועל של ה-DAG עשויה להתרחש בשעה 12:05, זמן מה אחרי שהתאריך הלוגי חלף.

מרווח הזמן שנקבע

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

לדוגמה, לוח זמנים יומי אומר ש-DAG מופעל פעם ביום, והתאריכים הלוגיים של הפעלות ה-DAG שלו הם במרווחים של 24 שעות.

תאריך התחלה

מגדיר מתי רוצים ש-Airflow יתחיל לתזמן את ה-DAG.

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

השלמה, מילוי חוסרים וניסיונות חוזרים

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

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

כך מתבצע התזמון:

  1. אחרי שתאריך ההתחלה חלף, Airflow מחכה להפעלה הבאה של מרווח התזמון.

  2. מערכת Airflow מתזמנת את ההרצה הראשונה של ה-DAG לסוף של מרווח הזמן הזה.

    לדוגמה, אם DAG מתוזמן להפעלה כל שעה ותאריך ההתחלה הוא 12:00 היום, ההפעלה הראשונה של ה-DAG תתרחש בשעה 13:00 היום.

בקטע תזמון של DAG ב-Airflow במסמך הזה מוסבר איך להגדיר תזמון ל-DAG באמצעות המושגים האלה. מידע נוסף על הפעלות ותזמון של DAG זמין במאמר DAG Runs (הפעלות של DAG) במאמרי העזרה של Airflow.

מידע על דרכים להפעלת DAG

ב-Airflow אפשר להפעיל DAG בדרכים הבאות:

  • הפעלה לפי לוח זמנים. ‫Airflow מפעיל את ה-DAG באופן אוטומטי על סמך התזמון שצוין עבורו בקובץ ה-DAG.

  • הפעלת הטריגר באופן ידני. אפשר להפעיל DAG באופן ידני מGoogle Cloud מסוף, מממשק המשתמש של Airflow או על ידי הפעלת פקודה של Airflow CLI מ-Google Cloud CLI.

  • טריגר בתגובה לאירועים. הדרך הרגילה להפעיל DAG בתגובה לאירועים היא שימוש בחיישן.

דרכים אחרות להפעלת DAG:

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

  • מוודאים שלחשבון שלכם יש תפקיד שמאפשר לנהל אובייקטים בקטגוריות של הסביבה, ולצפות ב-DAG ולהפעיל אותו. מידע נוסף מופיע במאמר בקרת גישה.

תזמון של DAG ב-Airflow

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

  1. מאתרים את קובץ ה-DAG במחשב ועורכים אותו. אם אין לכם את קובץ ה-DAG, אתם יכולים להוריד עותק שלו מקובץ ה-bucket של הסביבה. כשיוצרים קובץ DAG חדש, אפשר להגדיר את כל הפרמטרים.

  2. בפרמטר schedule, מגדירים את התזמון. אפשר להשתמש בביטוי Cron, כמו 0 0 * * *, או בהגדרה מראש, כמו @daily. מידע נוסף זמין במאמר Cron and Time Intervals (תזמון משימות באמצעות Cron ומרווחי זמן) במאמרי העזרה של Airflow.

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

  3. בפרמטר start_date, מגדירים את תאריך ההתחלה.

    מערכת Airflow קובעת את התאריך הלוגי של ההרצה הראשונה של DAG באמצעות הפרמטר הזה.

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

    התאריך הלוגי של הפעלות DAG שבוצעו במהלך ההשלמה יהיה תאריך בעבר, והתאריך של ההפעלה ישקף את הזמן שבו ההפעלה של ה-DAG בוצעה בפועל.

  5. (אופציונלי) בפרמטר retries מגדירים כמה פעמים Airflow צריך לנסות מחדש משימות שנכשלו (כל DAG מורכב ממשימה אחת או יותר). כברירת מחדל, המערכת מנסה לבצע מחדש משימות ב-Managed Airflow פעמיים.

  6. מעלים את הגרסה החדשה של ה-DAG לקטגוריה של הסביבה.

  7. מחכים עד ש-Airflow ינתח את ה-DAG בהצלחה. לדוגמה, אפשר לבדוק את רשימת ה-DAG בסביבה במסוףGoogle Cloud או בממשק המשתמש של Airflow.

הגדרת ה-DAG הבאה מופעלת פעמיים ביום, בשעות 00:00 ו-12:00. תאריך ההתחלה שלו מוגדר ל-1 בינואר 2024, אבל Airflow לא מפעיל אותו לתאריכים קודמים אחרי שמעלים או משהים אותו, כי ההפעלה של catchup מושבתת.

ה-DAG מכיל משימה אחת בשם insert_query_job, שמוסיפה שורה לטבלה באמצעות האופרטור BigQueryInsertJobOperator. האופרטור הזה הוא אחד מGoogle Cloud אופרטורים של BigQuery, שבהם אפשר להשתמש כדי לנהל מערכי נתונים וטבלאות, להריץ שאילתות ולאמת נתונים. אם ביצוע מסוים של המשימה הזו נכשל, Airflow מנסה לבצע אותה שוב ארבע פעמים נוספות עם מרווח ברירת המחדל בין הניסיונות. התאריך הלוגי של הניסיונות החוזרים האלה נשאר זהה.

שאילתת ה-SQL בשורה הזו משתמשת בתבניות Airflow כדי לכתוב את התאריך הלוגי והשם של ה-DAG בשורה.

import datetime

from airflow.models.dag import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG(
  "bq_example_scheduling_dag",
  start_date=datetime.datetime(2024, 1, 1),
  schedule='0 */12 * * *',
  catchup=False
  ) as dag:

  insert_query_job = BigQueryInsertJobOperator(
    task_id="insert_query_job",
    retries=4,
    configuration={
        "query": {
            # schema: date (string), description (string)
            # example row: "20240101T120000", "DAG run: <DAG: bq_example_scheduling_dag>"
            "query": "INSERT example_dataset.example_table VALUES ('{{ ts_nodash }}', 'DAG run: {{ dag }}' )",
            "useLegacySql": False,
            "priority": "BATCH",
        }
    },
    location="us-central1"
  )

  insert_query_job

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

דוגמאות נוספות לפרמטרים של תזמון

הדוגמאות הבאות לפרמטרים של תזמון ממחישות איך התזמון פועל עם שילובים שונים של פרמטרים:

  • אם start_date הוא datetime(2024, 4, 4, 16, 25) ו-schedule הוא 30 16 * * *, ההרצה הראשונה של ה-DAG תתבצע ב-5 באפריל 2024 בשעה 16:30.

  • אם start_date הוא datetime(2024, 4, 4, 16, 35) ו-schedule הוא 30 16 * * *, ההרצה הראשונה של ה-DAG תתבצע ב-6 באפריל 2024 בשעה 16:30. מכיוון שתאריך ההתחלה הוא אחרי מרווח התזמון ב-4 באפריל 2024, הפעלת ה-DAG לא מתרחשת ב-5 באפריל 2024. במקום זאת, מרווח הזמן של התזמון מסתיים בשעה 16:35 ב-5 באפריל 2024, ולכן ההפעלה הבאה של ה-DAG מתוזמנת לשעה 16:30 ביום המחרת.

  • אם start_date הוא datetime(2024, 4, 4), ו-schedule הוא @daily, ההרצה הראשונה של ה-DAG מתוזמנת לשעה 00:00 ב-5 באפריל 2024.

  • אם הערך של start_date הוא datetime(2024, 4, 4, 16, 30) והערך של schedule הוא 0 * * * *, ההרצה הראשונה של DAG מתוזמנת לשעה 18:00 ב-4 באפריל 2024. אחרי התאריך והשעה שצוינו, Airflow מתזמן הפעלה של DAG שתתבצע בדקה 0 של כל שעה. השעה הקרובה ביותר שבה זה קורה היא 17:00. בשלב הזה, מערכת Airflow מתזמנת הפעלה של DAG שתתרחש בסוף מרווח התזמון, כלומר בשעה 18:00.

הפעלת DAG באופן ידני

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

המסוף

כדי להפעיל DAG ממסוף Google Cloud :

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

    מעבר אל Environments

  2. בוחרים סביבה כדי לראות את הפרטים שלה.

  3. בדף פרטי הסביבה, עוברים לכרטיסייה DAGs.

  4. לוחצים על השם של DAG.

  5. בדף פרטי DAG, לוחצים על הפעלת DAG. נוצרת הפעלה חדשה של DAG.

    .

ממשק המשתמש של Airflow

כדי להפעיל DAG מממשק המשתמש של Airflow:

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

    מעבר אל Environments

  2. בעמודה Airflow webserver, לוחצים על הקישור Airflow של הסביבה.

  3. נכנסים לחשבון Google עם ההרשאות המתאימות.

  4. בממשק המשתמש של Airflow, בדף DAGs, לוחצים על הלחצן Trigger DAG (הפעלת DAG) ליד ה-DAG הרצוי.

gcloud

מריצים את הפקודה dags trigger Airflow CLI:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags trigger -- DAG_ID

מחליפים את מה שכתוב בשדות הבאים:

  • ‫ENVIRONMENT_NAME: השם של הסביבה.
  • ‫LOCATION: האזור שבו נמצאת הסביבה.
  • ‫DAG_ID: השם של ה-DAG.

מידע נוסף על הרצת פקודות Airflow CLI בסביבות Managed Airflow, ראו הרצת פקודות Airflow CLI.

למידע נוסף על הפקודות הזמינות של Airflow CLI, אפשר לעיין במדריך העזר לפקודה gcloud composer environments run.

צפייה ביומנים ובפרטים של הפעלת DAG

במסוף Google Cloud אפשר:

בנוסף, Managed Airflow מספק גישה לממשק המשתמש של Airflow, שהוא ממשק האינטרנט של Airflow.

השהיית DAG

המסוף

כדי להשהות DAG דרך Google Cloud המסוף:

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

    מעבר אל Environments

  2. בוחרים סביבה כדי לראות את הפרטים שלה.

  3. בדף פרטי הסביבה, עוברים לכרטיסייה DAGs.

  4. לוחצים על השם של DAG.

  5. בדף פרטי DAG, לוחצים על השהיית DAG.

ממשק המשתמש של Airflow

כדי להשהות DAG מממשק המשתמש של Airflow:

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

מעבר אל Environments

  1. בעמודה Airflow webserver, לוחצים על הקישור Airflow של הסביבה.

  2. נכנסים לחשבון Google עם ההרשאות המתאימות.

  3. בממשק האינטרנט של Airflow, בדף DAGs, לוחצים על המתג לצד שם ה-DAG.

gcloud

מריצים את הפקודה dags pause Airflow CLI:

  gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    dags pause -- DAG_ID

מחליפים את מה שכתוב בשדות הבאים:

  • ‫ENVIRONMENT_NAME: השם של הסביבה.
  • ‫LOCATION: האזור שבו נמצאת הסביבה.
  • ‫DAG_ID: השם של ה-DAG.

מידע נוסף על הרצת פקודות Airflow CLI בסביבות Managed Airflow, ראו הרצת פקודות Airflow CLI.

למידע נוסף על הפקודות הזמינות של Airflow CLI, אפשר לעיין במדריך העזר לפקודה gcloud composer environments run.

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