פתרון בעיות ב-DAG

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

בדף הזה מפורטים שלבים לפתרון בעיות ומידע על בעיות נפוצות בתהליכי עבודה.

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

פתרון בעיות באמצעות Managed Airflow Agent

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

תהליך עבודה לפתרון בעיות

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

1. בודקים את היומנים של Airflow.

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

   קטע	מפתח	ערך
   ‫logging (core ב-Airflow 1)	logging_level	ערך ברירת המחדל הוא INFO. מגדירים את הערך ל-DEBUG כדי לקבל רמת פירוט גבוהה יותר בהודעות היומן.

2. בודקים את לוח הבקרה של Monitoring.

3. בודקים את Cloud Monitoring.

4. במסוף Google Cloud , בודקים אם יש שגיאות בדפים של רכיבי הסביבה.

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

   קטע	מפתח	ערך
   webserver	dag_orientation	LR, TB, RL אוBT

ניפוי באגים בכשלים של אופרטורים

כדי לנפות באגים בכשל של אופרטור:

1. בודקים אם יש שגיאות שקשורות למשימה.
2. בודקים את היומנים של Airflow.
3. בודקים את Cloud Monitoring.
4. בודקים יומנים ספציפיים למפעיל.
5. מתקנים את השגיאות.
6. מעלים את ה-DAG לתיקייה /dags.
7. בממשק האינטרנט של Airflow, מנקים את המצבים הקודמים של ה-DAG.
8. ממשיכים או מפעילים את ה-DAG.

פתרון בעיות בביצוע משימות

‫Airflow היא מערכת מבוזרת עם הרבה ישויות כמו מתזמן, מפעיל, עובדים שתקשרים ביניהם דרך תור משימות ומסד הנתונים של Airflow, ושולחים אותות (כמו SIGTERM). התרשים הבא מציג סקירה כללית של הקישורים בין רכיבי Airflow.

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

משימות נכשלות בלי ליצור יומנים

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

שגיאות או פסק זמן בניתוח DAG

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

תסמינים

• יומני עובדים של Airflow ב-Cloud Logging מכילים הודעות כמו:

  • airflow.exceptions.AirflowException: Dag "example-dag" could not be found; either it does not exist or it failed to parse.
  • ‫airflow.exceptions.AirflowTaskTimeout: Timeout, PID: 12345 (יכול להיות שההודעה הזו לא תכיל את שם ה-DAG או את נתיב הקובץ).
  • ERROR - Failed to import: /home/airflow/gcs/dags/example-dag.py (ללא tracebacks מפורטים).

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

פתרון

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

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

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

• אם מופיעות שגיאות Failed to import או שגיאות בממשק המשתמש של Airflow או במסוףGoogle Cloud , כדאי לבדוק את היומנים של מעבד ה-DAG כדי לראות את פרטי ה-traceback, כמו שמתואר במאמר פתרון בעיות במעבד ה-DAG. אפשר גם להריץ את הפקודה הבאה ב-CLI של gcloud כדי לראות את השגיאות בייבוא ה-DAG:

  gcloud composer environments run ENVIRONMENT_NAME \
      --location LOCATION \
      dags list-import-errors
  

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

• אם אי אפשר לבצע אופטימיזציה של ניתוח DAG, צריך להגדיל את dagbag_import_timeout. עוקפים את אפשרות ההגדרה הזו של Airflow ומגדירים ערך גבוה יותר מ-30 שניות (ברירת המחדל), למשל 120 שניות.

עיכובים בסנכרון קובצי DAG

כשמעלים או מעדכנים קובצי DAG בדלי של הסביבה, לוקח זמן עד שהקבצים האלה מסתנכרנים עם העובדים והמתזמנים של Airflow. הסנכרון הזה מתבצע באופן עצמאי בכל המתזמנים והעובדים. אם מפעילים ריצת DAG זמן קצר אחרי העלאה או עדכון של קובץ DAG, והקובץ עדיין לא מסונכרן עם עובד שמבצע משימה, המשימה תיכשל ללא יומנים, ויכול להיות שיופיע airflow.exceptions.AirflowException: Dag "example-dag" could not be found... ביומני העובדים.

הסנכרון הזה בדרך כלל נמשך דקה או שתיים, אבל הוא יכול להימשך יותר זמן אם יש לכם הרבה קבצים או קבצים גדולים בתיקיות dags/ או plugins/ בדלי.

פתרון

צריך להמתין לפחות 2 דקות אחרי העלאה או עדכון של DAG או פלאגינים לפני שמפעילים DAG או מאפשרים אותם.

משימות תקועות במצב 'בתור'

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

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

כפתרון עקיף, אפשר לנקות ידנית משימות שנתקעו במצב 'בהמתנה'. בממשק המשתמש של Airflow, עוברים אל Browse > Task Instances, מוצאים מופעים של משימות שנתקעו במצב queued ומגדירים את המצב שלהם ל-failed.

המשימות מופסקות באופן פתאומי

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

משימות לא פעילות

מערכת Airflow מזהה שני סוגים של חוסר התאמה בין משימה לבין תהליך שמבצע את המשימה:

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

  איך לגלות משימות זומבי

  resource.type="cloud_composer_environment"
  resource.labels.environment_name="ENVIRONMENT_NAME"
  log_id("airflow-scheduler")
  textPayload:"Detected zombie job"

• משימות מתות הן משימות שלא אמורות לפעול. ‫Airflow מוצא משימות כאלה מעת לעת ומסיים אותן.

מידע נוסף על פתרון בעיות שקשורות למשימות זומבי זמין במאמר בנושא סיבות נפוצות לבעיות.

אותות SIGTERM

אותות SIGTERM משמשים את Linux,‏ Kubernetes, ‏ Airflow scheduler ו-Celery כדי להפסיק תהליכים שאחראים להפעלת עובדי Airflow או משימות Airflow.

יכולות להיות כמה סיבות לשליחת אותות SIGTERM בסביבה:

• משימה הפכה למשימת זומבי וצריך להפסיק אותה.

• מתזמן המשימות זיהה כפילות של משימה ושולח למשימה אותות של סיום המופע ו-SIGTERM כדי להפסיק אותה.

• בהתאמה אופקית של קבוצות Pod לעומס, מישור הבקרה של GKE שולח אותות SIGTERM כדי להסיר Pods שכבר לא נחוצים.

• מתזמן יכול לשלוח אותות SIGTERM לתהליך DagFileProcessorManager. האותות האלה של SIGTERM משמשים את Scheduler לניהול מחזור החיים של התהליך DagFileProcessorManager, ואפשר להתעלם מהם בבטחה.

  דוגמה:

  Launched DagFileProcessorManager with pid: 353002
  Sending Signals.SIGTERM to group 353002. PIDs of all processes in the group: []
  Sending the signal Signals.SIGTERM to group 353002
  Sending the signal Signals.SIGTERM to process 353002 as process group is missing.
  

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

  אפשר להתעלם בבטחה מאותות SIGTERM כאלה. המשימה כבר במצב מוצלח, והביצוע של הרצת ה-DAG כולה לא יושפע.

  רשומת היומן Received SIGTERM. היא ההבדל היחיד בין יציאה רגילה לבין סיום המשימה במצב מוצלח.

  

• רכיב Airflow משתמש ביותר משאבים (CPU, זיכרון) מהמותר על ידי צומת האשכול.

• שירות GKE מבצע פעולות תחזוקה ושולח אותות SIGTERM לקבוצות Pod שפועלות בצומת שעומד לעבור שדרוג.

  כשמופסקת פעילות של מופע של משימה באמצעות SIGTERM, אפשר לראות את רשומות היומן הבאות ביומנים של Airflow worker שהריץ את המשימה:

  {local_task_job.py:211} WARNING - State of this instance has been externally
  set to queued. Terminating instance. {taskinstance.py:1411} ERROR - Received
  SIGTERM. Terminating subprocesses. {taskinstance.py:1703} ERROR - Task failed
  with exception