testing-dags

DAG テストスキル

af コマンドを使用して、反復サイクルで DAG をテスト、デバッグ、修正します。

CLI の実行

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

---

Astro CLI によるクイック検証

ユーザーが Astro CLI を利用できる場合、これらのコマンドを使うと、実行中の Airflow インスタンスなしで迅速にフィードバックを得られます。

# Parse DAGs to catch import errors, syntax issues, and DAG-level problems
astro dev parse

# Run pytest against DAGs (runs tests in tests/ directory)
astro dev pytest

開発中のクイック検証に使用してください。稼働中の Airflow インスタンスに対してエンドツーエンドのテストを行う場合は、以下のトリガーして待機するワークフローに進んでください。

---

最初のアクション：DAG をトリガーするだけ

ユーザーから DAG のテストを依頼されたとき、最初に行うアクションはこれだけです。

af runs trigger-wait <dag_id>

次のことはしないでください：

• 最初に af dags list を呼び出す
• 最初に af dags get を呼び出す
• 最初に af dags errors を呼び出す
• grep や ls、その他の bash コマンドを使用する
• 「事前チェック」を行う

DAG をトリガーするだけです。 失敗した場合にのみ、デバッグしてください。

---

テストワークフローの概要

┌─────────────────────────────────────┐
│ 1. TRIGGER AND WAIT                 │
│    Run DAG, wait for completion     │
└─────────────────────────────────────┘
                 ↓
        ┌───────┴───────┐
        ↓               ↓
   ┌─────────┐    ┌──────────┐
   │ SUCCESS │    │ FAILED   │
   │ Done!   │    │ Debug... │
   └─────────┘    └──────────┘
                       ↓
        ┌─────────────────────────────────────┐
        │ 2. DEBUG (only if failed)           │
        │    Get logs, identify root cause    │
        └─────────────────────────────────────┘
                       ↓
        ┌─────────────────────────────────────┐
        │ 3. FIX AND RETEST                   │
        │    Apply fix, restart from step 1   │
        └─────────────────────────────────────┘

基本方針：まず試し、失敗したらデバッグします。 事前チェックに時間を費やさず、DAG を実行し、問題が起きた場合に診断してください。

---

フェーズ 1：トリガーして待機する

af runs trigger-wait を使用して DAG をテストします。

主な方法：トリガーして待機する

af runs trigger-wait <dag_id> --timeout 300

例：

af runs trigger-wait my_dag --timeout 300

この方法が推奨される理由：

• 1 つのコマンドでトリガーと監視を処理できます
• DAG の完了時（成功または失敗）にすぐに応答します
• 実行が失敗した場合、失敗したタスクの詳細が含まれます
• 手動でポーリングする必要はありません

応答の解釈

成功：

{
  "dag_run": {
    "dag_id": "my_dag",
    "dag_run_id": "manual__2025-01-14T...",
    "state": "success",
    "start_date": "...",
    "end_date": "..."
  },
  "timed_out": false,
  "elapsed_seconds": 45.2
}

失敗：

{
  "dag_run": {
    "state": "failed"
  },
  "timed_out": false,
  "elapsed_seconds": 30.1,
  "failed_tasks": [
    {
      "task_id": "extract_data",
      "state": "failed",
      "try_number": 2
    }
  ]
}

タイムアウト：

{
  "dag_id": "my_dag",
  "dag_run_id": "manual__...",
  "state": "running",
  "timed_out": true,
  "elapsed_seconds": 300.0,
  "message": "Timed out after 300 seconds. DAG run is still running."
}

代替方法：トリガーと監視を個別に行う

より細かく制御する必要がある場合にのみ使用してください。

# Step 1: Trigger
af runs trigger my_dag
# Returns: {"dag_run_id": "manual__...", "state": "queued"}

# Step 2: Check status
af runs get my_dag manual__2025-01-14T...
# Returns current state

---

結果への対応

成功した場合

DAG は正常に実行されました。次の内容をユーザーに要約してください。

• 合計経過時間
• 完了したタスクの数
• 特筆すべき出力（ログで確認できる場合）

これで完了です！

タイムアウトした場合

DAG は引き続き実行中です。次の方法があります。

1. 現在のステータスを確認する：af runs get <dag_id> <dag_run_id>
2. 待機を続けるかユーザーに確認する
3. タイムアウトを延長して再試行する

失敗した場合

フェーズ 2（デバッグ）に進み、根本原因を特定します。

---

フェーズ 2：失敗のデバッグ（必要な場合のみ）

DAG の実行が失敗した場合、次のコマンドで原因を診断します。

包括的な診断情報を取得する

af runs diagnose <dag_id> <dag_run_id>

1 回の呼び出しで次の情報が返されます。

• 実行メタデータ（状態、タイミング）
• すべてのタスクインスタンスとその状態
• 失敗したタスクの概要
• 状態の件数（成功、失敗、スキップなど）

タスクログを取得する

af tasks logs <dag_id> <dag_run_id> <task_id>

例：

af tasks logs my_dag manual__2025-01-14T... extract_data

特定の再試行に対して：

af tasks logs my_dag manual__2025-01-14T... extract_data --try 2

次の項目を確認してください：

• 例外メッセージとスタックトレース
• 接続エラー（データベース、API、S3）
• 権限エラー
• タイムアウトエラー
• 不足している依存関係

上流タスクを確認する

タスクに upstream_failed と表示されている場合、根本原因は上流タスクにあります。af runs diagnose を使用して、実際に失敗したタスクを特定してください。

インポートエラーを確認する（DAG が実行されなかった場合）

DAG が存在しないためトリガーに失敗した場合：

af dags errors

これにより、DAG の読み込みを妨げた構文エラーや不足している依存関係が明らかになります。

---

フェーズ 3：修正して再テストする

問題を特定したら、次の手順に進みます。

よくある修正

問題	修正
インポートの不足	DAG ファイルに追加する
パッケージの不足	requirements.txt に追加する
接続エラー	af config connections を確認し、認証情報を検証する
変数の不足	af config variables を確認し、必要に応じて作成する
タイムアウト	タスクのタイムアウトを延長するか、クエリを最適化する
権限エラー	接続の認証情報を確認する

修正後

1. ファイルを保存する
2. 再テストする： af runs trigger-wait <dag_id>

DAG が成功するまで、テスト → デバッグ → 修正のサイクルを繰り返してください。

---

CLI クイックリファレンス

フェーズ	コマンド	目的
テスト	af runs trigger-wait <dag_id>	主なテスト方法 — まずはこちらを実行してください
テスト	af runs trigger <dag_id>	実行を開始します（代替手段）
テスト	af runs get <dag_id> <run_id>	実行のステータスを確認します
デバッグ	af runs diagnose <dag_id> <run_id>	失敗の包括的な診断
デバッグ	af tasks logs <dag_id> <run_id> <task_id>	タスクの出力／エラーを取得します
デバッグ	af dags errors	解析エラーを確認します（DAG が読み込まれない場合）
デバッグ	af dags get <dag_id>	DAG の設定を確認します
デバッグ	af dags explore <dag_id>	DAG を詳細に検査します
設定	af config connections	コネクションの一覧を表示します
設定	af config variables	変数の一覧を表示します

---

テストシナリオ

シナリオ 1：DAG のテスト（正常系）

af runs trigger-wait my_dag
# Success! Done.

シナリオ 2：DAG のテスト（失敗を伴う場合）

# 1. Run and wait
af runs trigger-wait my_dag
# Failed...

# 2. Find failed tasks
af runs diagnose my_dag manual__2025-01-14T...

# 3. Get error details
af tasks logs my_dag manual__2025-01-14T... extract_data

# 4. [Fix the issue in DAG code]

# 5. Retest
af runs trigger-wait my_dag

シナリオ 3：DAG が存在しない／読み込まれない場合

# 1. Trigger fails - DAG not found
af runs trigger-wait my_dag
# Error: DAG not found

# 2. Find parse error
af dags errors

# 3. [Fix the issue in DAG code]

# 4. Retest
af runs trigger-wait my_dag

シナリオ 4：スケジュール実行の失敗をデバッグする

# 1. Get failure summary
af runs diagnose my_dag scheduled__2025-01-14T...

# 2. Get error from failed task
af tasks logs my_dag scheduled__2025-01-14T... failed_task_id

# 3. [Fix the issue]

# 4. Retest
af runs trigger-wait my_dag

シナリオ 5：カスタム設定でテストする

af runs trigger-wait my_dag --conf '{"env": "staging", "batch_size": 100}' --timeout 600

シナリオ 6：長時間実行される DAG

# Wait up to 1 hour
af runs trigger-wait my_dag --timeout 3600

# If timed out, check current state
af runs get my_dag manual__2025-01-14T...

---

デバッグのヒント

よくあるエラーパターン

接続拒否／タイムアウト：

• af config connections でホスト／ポートが正しいか確認します
• 外部システムへのネットワーク接続を確認します
• コネクションの認証情報が正しいか確認します

ModuleNotFoundError：

• requirements.txt にパッケージがありません
• 追加後、環境の再起動が必要な場合があります

PermissionError：

• IAM ロール、データベースの権限付与、API キーを確認します
• コネクションに正しい認証情報が設定されているか確認します

タスクのタイムアウト：

• クエリまたは処理に時間がかかりすぎています
• タスクにタイムアウトパラメーターを追加することを検討します
• 基盤となるクエリ／処理を最適化します

タスクログの読み方

タスクログには通常、次の内容が表示されます。

1. タスクの開始タイムスタンプ
2. タスクコード内の print／ログ出力
3. 戻り値（@task でデコレートされた関数の場合）
4. 例外とスタックトレース全体（失敗した場合）
5. タスクの終了タイムスタンプと所要時間

タスクが失敗した場合は、ログの末尾にある例外に注目してください。

Astro の場合

Astro のデプロイでは環境昇格がサポートされており、テストのワークフローを体系化するのに役立ちます。

• 開発デプロイ：astro deploy --dags を使って DAG を自由にテストし、素早く反復します
• ステージングデプロイ：本番環境に近いデータで統合テストを実行します
• 本番デプロイ：下位環境で検証してからデプロイします
• 環境ごとに個別の Astro デプロイを使用し、それらを通じてコードを昇格させます

---

関連スキル

• authoring-dags：新しい DAG を作成する場合（テスト前の検証を含む）
• debugging-dags：Airflow の一般的なトラブルシューティング
• deploying-airflow：テスト後に DAG を本番環境へデプロイする場合