debugging-dags

DAG の診断

Airflow DAG の失敗をデバッグするデータエンジニアとして、以下の体系的なアプローチに従い、根本原因を特定して、実行可能な修正策を提示してください。

CLI の実行

以下のコマンドは、af が PATH 上にあることを前提としています。astro otto 経由で実行すれば自動的に利用できます。または、uv tool install astro-airflow-mcp でスタンドアロン版をインストールできます。

---

ステップ 1：失敗を特定する

特定の DAG が指定されている場合：

• af runs diagnose <dag_id> <dag_run_id> を実行します（run_id が指定されている場合）
• run_id が指定されていない場合は、af dags stats を実行して最近の失敗を確認します

DAG が指定されていない場合：

• af health を実行して、すべての DAG の最近の失敗を確認します
• af dags errors でインポートエラーを確認します
• 最近失敗した DAG を表示します
• さらに調査する DAG を尋ねます

ステップ 2：エラーの詳細を確認する

失敗したタスクを特定したら：

1. af tasks logs <dag_id> <dag_run_id> <task_id> を使ってタスクログを取得します
2. 実際の例外を探します — Airflow の定型ログを読み進め、実際のエラーを見つけます
3. 失敗の種類を分類します：
   • データの問題：データの欠落、スキーマの変更、null 値、制約違反
   • コードの問題：バグ、構文エラー、インポートの失敗、型エラー
   • インフラストラクチャの問題：接続タイムアウト、リソース枯渇、権限不足
   • 依存関係の問題：上流の失敗、外部 API の停止、レート制限

ステップ 3：コンテキストを確認する

原因を理解するため、追加のコンテキストを収集します。

1. 最近の変更：コードがデプロイされましたか？利用できる場合は git の履歴を確認します
2. パッケージのバージョン変更：パッケージがアップグレードされましたか — イメージ内、venv 形式のオペレーター内、またはインデックス上での変更を確認します。詳しくは以下のパッケージのバージョン変更を参照してください。
3. データ量：データ量が急増しましたか？ソーステーブルの件数を簡単に確認します
4. 上流の健全性：上流のタスクは成功したものの、予期しないデータを生成していませんか？
5. 過去のパターン：同じ失敗が繰り返されていますか？同じタスクが以前にも失敗したか確認します
6. タイミング：通常と異なる時間に失敗しましたか？（リソースの競合、メンテナンス時間帯など）

af runs get <dag_id> <dag_run_id> を使って、失敗した実行を最近成功した実行と比較します。

パッケージのバージョン変更

Git の変更がないのに失敗する一般的な原因は、依存関係のドリフトです。ユーザーのコードは変わっていなくても、依存しているパッケージが変わっている可能性があります。次の順に確認してください。

1. ワーカーイメージの差分（利用できる場合は優先）。Astro のデプロイごとに新しいイメージタグが付くため、レジストリには「変更前」と「変更後」のイメージがあります。現在のイメージと前のイメージの pip freeze の差分を取ってください。これが実際に何が変わったかを示す確実な情報です。

   docker run --rm <current_image> pip freeze > /tmp/now.txt
   docker run --rm <previous_image> pip freeze > /tmp/prev.txt
   diff /tmp/prev.txt /tmp/now.txt
   

   2つのイメージで docker run --rm <image> python --version の結果も比較してください。pip freeze が同じに見える場合でも、Python のマイナーバージョンの更新（3.11 → 3.12、あるいはパッチバージョンの更新でも）により、wheel の互換性が損なわれることがあります。af config providers には現在インストールされているプロバイダーのバージョンが表示されるため、トレースバックに記載されたモジュールとの照合に役立ちます。

2. venv スタイルのオペレーターはワーカーイメージを経由しません。 @task.virtualenv、PythonVirtualenvOperator、ExternalPythonOperator、KubernetesPodOperator はタスクの実行ごとに環境を構築するため、イメージの差分では内部の失敗を検出できません。失敗したタスクがこれらのいずれかであれば、そのタスクの requirements／image／python_version／python 引数を直接確認してください。

   • 上限のない指定子（例：上限を指定していない pandas>=2.0.0、または指定子なし）→ 上流での新しいリリースが最有力の原因です。
   • image="foo:latest" またはタグなし → 気づかないうちにイメージが更新されています。
   • python_version="3.11"（@task.virtualenv／PythonVirtualenvOperator の場合）、または以前とは異なるインタープリターに解決される python パス（ExternalPythonOperator の場合）→ Python のマイナーバージョンの変更により、requirements が変わっていなくても wheel の互換性が損なわれることがあります。ベースの Python が変わった場合、ワーカーイメージ自体にも同じことが当てはまります。

   解決策はバージョンを固定することです。例：pandas>=2.0.0,<3.0.0、ロックファイル、特定のイメージ SHA、または完全な Python バージョン（"3.11" ではなく python_version="3.11.7"）。

3. インデックスの検索。イメージの差分で結論が出ない場合（イメージの履歴がない場合、または venv スタイルのオペレーターの場合）に行います。まず、設定されているインデックスを特定してください。PyPI とは限りません。

   • 環境変数：UV_INDEX_URL、PIP_INDEX_URL、PIP_EXTRA_INDEX_URL
   • pyproject.toml → [[tool.uv.index]]
   • ~/.pip/pip.conf、/etc/pip.conf
   • Dockerfile の --index-url フラグ

   次に、最初の失敗が発生した時点以降に、原因と疑われるパッケージのリリースがあったかを検索してください。PyPI の場合：

   curl -s https://pypi.org/pypi/<pkg>/json | jq '.releases | to_entries | map({version: .key, uploaded: .value[0].upload_time}) | sort_by(.uploaded) | reverse | .[:5]'
   

   プライベートインデックスでは通常、同じ /pypi/<pkg>/json 形式を利用できます。利用できない場合は Simple API（/simple/<pkg>/）を試すか、どちらも機能しない場合はユーザーに確認してください。

トレースバックに記載されたパッケージのリリース日時が、最後に成功した実行と最初に失敗した実行の間にあれば、それが原因です。

Astro の場合

Astro を実行している場合、診断には次のツールも役立ちます。

• デプロイアクティビティログ：Astro UI で最近のデプロイを確認してください。デプロイの失敗や最近のコード変更が、突然の失敗の原因であることはよくあります。
• Astro アラート：Astro UI でアラートを設定し、失敗（DAG の失敗、タスクの実行時間、SLA の未達）を事前に監視してください。
• オブザーバビリティ：Astro のオブザーバビリティダッシュボードを使用して、DAG の健全性の傾向を追跡し、繰り返し発生する問題を見つけてください。

OSS Airflow の場合

• Airflow UI：DAGs ページ、Graph ビュー、タスクログを使用して、最近の実行と失敗を調査してください。

ステップ 4：実行可能な出力を提示する

診断結果は次の形式でまとめてください。

根本原因

実際に何が壊れたのでしょうか。「タスクが失敗した」ではなく、「コードでは 0％を想定していた列 X が、行の 15％で null だったためタスクが失敗した」のように具体的に記述してください。

影響の評価

• どのデータが影響を受けていますか？更新されなかったテーブルはどれですか？
• どの下流プロセスがブロックされていますか？
• 本番環境のダッシュボードやレポートに影響していますか？

即時の修正

今すぐ解決するための具体的な手順：

1. データの問題の場合：不正なレコードを修正またはスキップする SQL
2. コードの問題の場合：必要なコード変更の詳細
3. インフラの問題の場合：連絡先、または再起動するもの

防止策

再発を防ぐ方法：

• データ品質チェックを追加しますか？
• エラー処理を改善しますか？
• エッジケースのアラートを追加しますか？
• ドキュメントを更新しますか？
• 上流の変更が気付かないうちに反映されるのを避けるため、依存関係を固定しますか（制約ファイル、ロックファイル、または venv／external／pod オペレーターの上限指定子を使用）？

クイックコマンド

すぐに使えるコマンドを示します：

• DAG の実行全体をクリアして再実行するには：af runs clear <dag_id> <run_id>
• 失敗した特定のタスクをクリアして再実行するには：af tasks clear <dag_id> <run_id> <task_ids> -D
• 停滞している、または不要な実行を削除するには：af runs delete <dag_id> <run_id>