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

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

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

• ל-API של ניתוח שיחות שמופעל על ידי Managed Airflow Agent יש מדיניות משלו בנושא מיקום אחסון הנתונים. יכול להיות שהשיחה שלכם תטופל באזור שונה מהמיקום של הסביבה שלכם.

בדיקת משימות Airflow שנכשלו או הפעלות DAG שנכשלו

כדי לחקור משימת Airflow שנכשלה או הרצה של DAG:

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

   מעבר אל Environments

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

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

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

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

6. איתור של הרצת DAG שנכשלה. אתם יכולים לבדוק הפעלה של DAG שלם או משימה ספציפית שנכשלה:

   • כדי לחקור הפעלה של DAG:

     1. לוחצים על auto_awesome חקירה לצד השם של הפעלת ה-DAG שנכשלה.

     2. בתיבת הדו-שיח Investigate DAG run failure (בדיקת כשל בהרצת DAG), לוחצים על Investigate (בדיקה).

   • כדי לחקור משימה ב-Airflow:

     1. לוחצים על השם של הרצת ה-DAG שנכשלה כדי להציג רשימה של משימות.

     2. בעמודה State של משימת Airflow שנכשלה, לוחצים על auto_awesome Investigate.

     3. בתיבת הדו-שיח Investigate task failure (חקירת כשל במשימה), לוחצים על Investigate (חקירה).

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

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

דוגמה לחקירה

בדוגמה הזו מוצג תהליך של ניתוח משימה שנכשלה.

1. במרכז הבקרה DAG Statistics (נתונים סטטיסטיים של DAG) בדף Monitoring (מעקב) >, בודקים את הנתונים של הפעלות DAG שנכשלו:

   

2. עוברים לכרטיסייה DAGs (תרשימי זרימה מכווני מחזורים). בעמודה Failed runs (1h) (הפעלות שנכשלו (שעה)) אפשר לראות של-DAG‏ create_large_txt_file_print_logs היו כמה הפעלות שנכשלו בשעה האחרונה. לוחצים על השם של ה-DAG.

   

3. לוחצים על אחת מהרצות ה-DAG שנכשלו ואז לוחצים על auto_awesome Investigate (חקירה) לצד הרשומה של משימת Airflow שנכשלה.

   

4. מחכים עד שהסוכן יסיים את הדוח.

5. בודקים את דוח האבחון. בדוגמה הזו, המשימה נכשלה בלי ליצור יומנים, אבל Managed Airflow Agent הצליח למצוא את הסיבה לכישלון ביומנים של מתזמן Airflow, שבהם המשימה הופסקה כזומבי:

   ### **1. Problem**
   
   The task `create_huge_file` in DAG `dag_simulation` failed because the
   Airflow worker executing it ran out of memory (OOM). This caused the task
   container or process to be abruptly terminated, preventing it from sending
   heartbeats to the scheduler. As a result, the scheduler detected it as a
   zombie task and marked it as failed.
   
   ### **2. Evidence**
   
   *  **Scheduler Log (Zombie Detection)**:
      Detected a task instance without a heartbeat:
      filepath='dag-simulation.py'
      bundle_name='dags-folder' bundle_version=None msg="{'DAG Id':
      'dag_simulation', 'Task Id': 'create_huge_file', 'Run Id':
      'manual__2026-11-25T12:42:50.190250+00:00', 'Hostname':
      'airflow-worker-2lxm2', 'External Executor Id':
      'cfc98619-0de6-4699-96e6-7ba8a5a581aa'}"
   
   *  ...
   

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

   ### **3. Recommendation**
   To resolve this issue, you must prevent the task from consuming unbound
   memory. Depending on your workflow:
   
   *  **Fix Code Logic (Highly Recommended)**: Avoid loading large datasets
      completely into worker memory. Implement chunk-based processing or
      streaming, or set up explicit safety limits. For example, if you need to
      limit memory allocation or process in bounded batches, do not keep
      references to old chunks in memory: ...
   
   *  **Scale Worker Memory**: If processing massive datasets in memory is
      unavoidable, consider upgrading the resource limits (RAM/CPU) of your
      Airflow worker pools or offloading the heavy processing step to external
      services.
   

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

• פתרון בעיות ב-DAG
• פתרון בעיות בתזמון
• פתרון בעיות בהתקנת חבילות PyPI