Managed Airflow エージェントを使用して Airflow タスクと DAG 実行のトラブルシューティングを行う

Managed Airflow エージェントは、失敗した Airflow タスクと DAG 実行の問題を把握、診断、解決するのに役立ちます。

始める前に

• Managed Airflow エージェントを実行する Conversational Analytics API には、 独自のデータ所在地ポリシーがあります。会話は、環境のロケーションとは異なるリージョンで処理される場合があります。

失敗した Airflow タスクまたは DAG 実行を調査する

失敗した Airflow タスクまたは DAG 実行を調査するには:

1. コンソール Cloud de Confiance で、[Environments] ページに移動します。

   [環境] に移動

2. 環境を選択して詳細を表示します。

3. [環境の詳細] ページで、[DAG] タブに移動します。

4. DAG の名前をクリックします。

5. [DAG の詳細] ページで、[実行履歴] タブに移動します。

6. 失敗した DAG 実行を見つけます。DAG 実行全体または特定の失敗したタスクを調査できます。

   • DAG 実行を調査するには:

     1. 失敗した DAG 実行名の横にある auto_awesome [Investigate] をクリックします。

     2. [DAG 実行失敗の調査] ダイアログで、[調査] をクリックします。

   • Airflow タスクを調査するには:

     1. 失敗した DAG 実行の名前をクリックして、タスクのリストを表示します。

     2. 失敗した Airflow タスクの [状態] 列で、 auto_awesome [調査] をクリックします。

     3. [タスクの失敗を調査] ダイアログで、[調査] をクリックします。

7. [エージェントに質問] パネルが開き、 Managed Airflow エージェントが DAG 実行の 失敗の調査を開始します。エージェントが分析を完了するまで待ってから、 診断レポートを確認します。診断レポートはいくつかのセクションで構成されています。

   • [問題] には、問題の根本原因が記載されています。
   • [証拠] は、エージェントの結論を裏付ける特定のアーティファクトを示しています。
   • [推奨事項] には、問題を解決するための具体的な手順が記載されています。

調査の例

この例では、失敗したタスクを分析するプロセスを示します。

1. [モニタリング] > [DAG の統計情報] ダッシュボードで、 失敗した DAG 実行を確認します。

   

2. [DAG] タブに移動します。[失敗した実行（1 時間）] 列には、過去 1 時間に create_large_txt_file_print_logs DAG が複数回失敗したことが示されています。DAG の名前をクリックします。

   

3. 失敗した DAG 実行のいずれかをクリックし、失敗した Airflow タスク エントリの横にある auto_awesome [**調査**]をクリックします。

   

4. エージェントがレポートを完了するまで待ちます。

5. 診断レポートを確認します。この例では、タスクはログを生成せずに失敗しましたが、Managed Airflow エージェントは、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 パッケージのインストールのトラブルシューティング