airflow

Airflow の運用

af コマンドを使用して、Airflow ワークフローの照会、管理、トラブルシューティングを行います。

Astro CLI

Astro CLI は、Airflow をローカルで実行し、本番環境にデプロイするための推奨方法です。すぐに使えるコンテナ化された Airflow 環境を提供します。

# Initialize a new project
astro dev init

# Start local Airflow (webserver at http://localhost:8080)
astro dev start

# Parse DAGs to catch errors quickly (no need to start Airflow)
astro dev parse

# Run pytest against your DAGs
astro dev pytest

# Deploy to production
astro deploy            # Full deploy (image + DAGs)
astro deploy --dags     # DAG-only deploy (fast, no image build)

詳しくは、以下をご覧ください。

• 新しいプロジェクトを始めますか？ setting-up-astro-project スキルをご覧ください
• ローカル環境を構築しますか？ managing-astro-local-env スキルをご覧ください
• デプロイしますか？ deploying-airflow スキルをご覧ください

---

CLI の実行

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

インスタンス設定

永続的な設定を使用して、複数の Airflow インスタンスを管理します。

# Add a new instance
af instance add prod --url https://airflow.example.com --token "$API_TOKEN"
af instance add staging --url https://staging.example.com --username admin --password admin

# List and switch instances
af instance list      # Shows all instances in a table
af instance use prod  # Switch to prod instance
af instance current   # Show current instance
af instance delete old-instance

# Auto-discover instances (use --dry-run to preview first)
af instance discover --dry-run        # Preview all discoverable instances
af instance discover                  # Discover from all backends (astro, local)
af instance discover astro            # Discover Astro deployments only
af instance discover astro --all-workspaces  # Include all accessible workspaces
af instance discover local            # Scan common local Airflow ports
af instance discover local --scan     # Deep scan all ports 1024-65535

# IMPORTANT: Always run with --dry-run first and ask for user consent before
# running discover without it. The non-dry-run mode creates API tokens in
# Astro Cloud, which is a sensitive action that requires explicit approval.

# Show where an instance came from (file path + scope)
af instance show prod

# Override instance for a single command via env vars
AIRFLOW_API_URL=https://staging.example.com AIRFLOW_AUTH_TOKEN=$STG af dags list

# Or switch persistently
af instance use staging

設定のレイアウト（git config の system/global/local と同様です）：

スコープ	ファイル	コミット対象？
グローバル	~/.astro/config.yaml	該当なし（ユーザーごと）
プロジェクト共有	<root>/.astro/config.yaml	はい
プロジェクトローカル	<root>/.astro/config.local.yaml	いいえ（gitignore 対象）

<root> は、cwd から上位にたどり、.astro/ を探して決定されます。プロジェクト内でのデフォルトの書き込み先は、add／discover → プロジェクト共有、use → プロジェクトローカルです。--global／--project／--local で上書きできます。AF_CONFIG=<path> を設定すると、レイヤー構成を迂回して単一のファイルを使用します。

af migrate を使って、従来の ~/.af/config.yaml から移行できます（冪等です。古いファイルの名前を .bak に変更します）。

設定内のトークンでは、${VAR} 構文を使って環境変数を参照できます。

instances:
- name: prod
  url: https://airflow.example.com
  auth:
    token: ${AIRFLOW_API_TOKEN}

または、環境変数を直接使用することもできます（設定ファイルは不要です）。

export AIRFLOW_API_URL=http://localhost:8080
export AIRFLOW_AUTH_TOKEN=your-token-here
# Or username/password:
export AIRFLOW_USERNAME=admin
export AIRFLOW_PASSWORD=admin

あるいは、CLI フラグを使います：af --airflow-url http://localhost:8080 --token "$TOKEN" <command>

クイックリファレンス

コマンド	説明
af health	システムのヘルスチェック
af dags list	すべての DAG を一覧表示
af dags get <dag_id>	DAG の詳細を取得
af dags explore <dag_id>	DAG を詳細に調査
af dags source <dag_id>	DAG のソースコードを取得
af dags pause <dag_id>	DAG のスケジュールを一時停止
af dags unpause <dag_id>	DAG のスケジュールを再開
af dags errors	インポートエラーを一覧表示
af dags warnings	DAG の警告を一覧表示
af dags stats	DAG 実行の統計情報
af runs list	DAG 実行を一覧表示
af runs get <dag_id> <run_id>	実行の詳細を取得
af runs trigger <dag_id>	DAG の実行をトリガー
af runs trigger-wait <dag_id>	DAG の実行をトリガーし、完了を待機
af runs delete <dag_id> <run_id>	DAG 実行を完全に削除
af runs clear <dag_id> <run_id>	再実行のために実行状態をクリア
af runs diagnose <dag_id> <run_id>	失敗した実行を診断
af tasks list <dag_id>	DAG 内のタスクを一覧表示
af tasks get <dag_id> <task_id>	タスク定義を取得
af tasks instance <dag_id> <run_id> <task_id>	タスクインスタンスを取得
af tasks logs <dag_id> <run_id> <task_id>	タスクログを取得
af config version	Airflow のバージョン
af config show	設定全体を表示
af config connections	接続を一覧表示
af config variables	変数を一覧表示
af config variable <key>	指定した変数を取得
af config pools	プールを一覧表示
af config pool <name>	プールの詳細を取得
af config plugins	プラグインを一覧表示
af config providers	プロバイダーを一覧表示
af config assets	アセット／データセットを一覧表示
af api <endpoint>	REST API に直接アクセス
af api ls	利用可能な API エンドポイントを一覧表示
af api ls --filter X	パターンに一致するエンドポイントを一覧表示
af registry providers	Airflow Registry 内のプロバイダーを一覧表示
af registry modules <provider>	プロバイダー内のオペレーター／フック／センサー／転送機能を一覧表示
af registry parameters <provider>	プロバイダーのクラスのコンストラクターシグネチャ（名前、型、デフォルト値、必須かどうか）
af registry connections <provider>	プロバイダーが公開する接続タイプ

ユーザーの意図パターン

はじめに

• 「Airflow をローカルで実行するには？」／「Airflow をセットアップする」→ managing-astro-local-env スキルを使用します（Astro CLI を使用）
• 「新しい Airflow プロジェクトを作成する」／「プロジェクトを初期化する」→ setting-up-astro-project スキルを使用します（Astro CLI を使用）
• 「Airflow をインストールするには？」／「Airflow を始める」→ setting-up-astro-project スキルを使用します

DAG の操作

• 「どの DAG がありますか？」／「すべての DAG を一覧表示する」→ af dags list
• 「DAG X について教えて」／「DAG Y とは？」→ af dags explore <dag_id>
• 「DAG X のスケジュールは？」→ af dags get <dag_id>
• 「DAG X のコードを表示して」→ af dags source <dag_id>
• 「DAG X を停止して」／「このワークフローを一時停止して」→ af dags pause <dag_id>
• 「DAG X を再開して」→ af dags unpause <dag_id>
• 「DAG のエラーはありますか？」→ af dags errors
• 「新しい DAG を作成する」／「パイプラインを書く」→ authoring-dags スキルを使用します

実行の操作

• 「どの実行が行われましたか？」→ af runs list
• 「DAG X を実行して」／「パイプラインをトリガーして」→ af runs trigger <dag_id>
• 「DAG X を実行して、完了まで待って」→ af runs trigger-wait <dag_id>
• 「この実行が失敗した理由は？」→ af runs diagnose <dag_id> <run_id>
• 「この実行を削除して」／「停止した実行を削除して」→ af runs delete <dag_id> <run_id>
• 「この実行をクリアして」／「この実行を再試行して」／「これを再実行して」→ af runs clear <dag_id> <run_id>
• 「この DAG をテストして、失敗したら修正して」→ testing-dags スキルを使用します

タスクの操作

• 「DAG X にはどのタスクがありますか？」→ af tasks list <dag_id>
• 「タスクログを取得して」／「タスクが失敗した理由は？」→ af tasks logs <dag_id> <run_id> <task_id>
• 「根本原因を徹底的に分析して」／「診断して修正して」→ debugging-dags スキルを使用します

データの操作

• 「データは最新ですか？」／「このテーブルが最後に更新されたのはいつですか？」→ checking-freshness スキルを使用します
• 「このデータの取得元は？」→ tracing-upstream-lineage スキルを使用します
• 「このテーブルに依存しているものは？」／「これを変更すると何が壊れますか？」→ tracing-downstream-lineage スキルを使用します

デプロイの操作

• 「DAG をデプロイして」／「本番環境にプッシュして」→ deploying-airflow スキルを使用します
• 「CI/CD をセットアップして」／「デプロイを自動化して」→ deploying-airflow スキルを使用します
• 「Kubernetes にデプロイして」／「Helm をセットアップして」→ deploying-airflow スキルを使用します
• 「astro deploy」／「DAG のみをデプロイ」→ deploying-airflow スキルを使用します

システムの操作

• 「Airflow のバージョンは？」→ af config version
• 「どの接続がありますか？」→ af config connections
• 「プールはいっぱいですか？」→ af config pools
• 「Airflow は正常ですか？」→ af health

API の調査

• 「どの API エンドポイントが利用できますか？」→ af api ls
• 「変数のエンドポイントを検索して」→ af api ls --filter variable
• 「XCom の値にアクセスして」／「XCom を取得して」→ af api xcom-entries -F dag_id=X -F task_id=Y
• 「イベントログを取得して」／「監査証跡」→ af api event-logs -F dag_id=X
• 「API 経由で接続を作成して」→ af api connections -X POST --body '{...}'
• 「API 経由で変数を作成して」→ af api variables -X POST -F key=name -f value=val

レジストリの検索

• 「プロバイダー X にはどのオペレーターがありますか？」→ af registry modules <provider>
• 「オペレーター Y のコンストラクターのパラメーターは？」→ af registry parameters <provider>
• 「どのプロバイダーがありますか？」／「Z 用のプロバイダーはありますか？」→ af registry providers
• 「プロバイダー X が公開している接続タイプは？」→ af registry connections <provider>
• 「特定のオペレーターを使って DAG を作成する」→ 例をコピーする前に、レジストリで現在のシグネチャを確認します

よくあるワークフロー

デプロイ前に DAG を検証する

Astro CLI を使用している場合は、Airflow インスタンスを起動せずに DAG を検証できます。

# Parse DAGs to catch import errors and syntax issues
astro dev parse

# Run unit tests
astro dev pytest

それ以外の場合は、実行中のインスタンスに対して検証します。

af dags errors     # Check for parse/import errors
af dags warnings   # Check for deprecation warnings

コードを書く前にオペレーターのシグネチャを確認する

Airflow Registry（airflow.apache.org/registry）は、プロバイダークラスと現在のコンストラクターシグネチャに関する信頼できる情報源です。DAG を作成する際は、記憶や古いドキュメントではなく、こちらを優先してください。レジストリには、プロバイダーの最新リリースが反映されています。

# List all providers and pick the one you need
af registry providers | jq '.providers[] | {id, name, version}'

# List every operator / hook / sensor in a provider (e.g. standard, amazon, google)
af registry modules standard \
  | jq '.modules[] | {name, type, import_path, docs_url}'

# Get the current constructor signature for a specific class
af registry parameters standard \
  | jq '.classes["airflow.providers.standard.operators.hitl.ApprovalOperator"].parameters'

# Filter modules by substring (useful when you know the concept but not the class)
af registry modules standard \
  | jq '.modules[] | select(.import_path | test("hitl"))'

結果はローカルにキャッシュされます。最新バージョンは 1 時間、固定バージョン（変更されません）は 30 日間です。特定のリリースを対象にするには、任意の modules ／ parameters ／ connections 呼び出しに --version X.Y.Z を追加します。

失敗した実行を調査する

# 1. List recent runs to find failure
af runs list --dag-id my_dag

# 2. Diagnose the specific run
af runs diagnose my_dag manual__2024-01-15T10:00:00+00:00

# 3. Get logs for failed task (from diagnose output)
af tasks logs my_dag manual__2024-01-15T10:00:00+00:00 extract_data

# 4. After fixing, clear the run to retry all tasks
af runs clear my_dag manual__2024-01-15T10:00:00+00:00

朝のヘルスチェック

# 1. Overall system health
af health

# 2. Check for broken DAGs
af dags errors

# 3. Check pool utilization
af config pools

DAG を理解する

# Get comprehensive overview (metadata + tasks + source)
af dags explore my_dag

DAG が実行されない理由を確認する

# Check if paused
af dags get my_dag

# Check for import errors
af dags errors

# Check recent runs
af runs list --dag-id my_dag

トリガーして監視する

# Option 1: Trigger and wait (blocking)
af runs trigger-wait my_dag --timeout 1800

# Option 2: Trigger and check later
af runs trigger my_dag
af runs get my_dag <run_id>

出力形式

instance コマンド（人が読みやすい表を使用）を除き、すべてのコマンドは JSON を出力します。

af dags list
# {
#   "total_dags": 5,
#   "returned_count": 5,
#   "dags": [...]
# }

フィルタリングには jq を使用します。

# Find failed runs
af runs list | jq '.dag_runs[] | select(.state == "failed")'

# Get DAG IDs only
af dags list | jq '.dags[].dag_id'

# Find paused DAGs
af dags list | jq '[.dags[] | select(.is_paused == true)]'

タスクログのオプション

# Get logs for specific retry attempt
af tasks logs my_dag run_id task_id --try 2

# Get logs for mapped task index
af tasks logs my_dag run_id task_id --map-index 5

af api による直接 API アクセス

高レベルコマンドでカバーされていないエンドポイント（XCom、event-logs、backfills など）には、af api を使用します。

# Discover available endpoints
af api ls
af api ls --filter variable

# Basic usage
af api dags
af api dags -F limit=10 -F only_active=true
af api variables -X POST -F key=my_var -f value="my value"
af api variables/old_var -X DELETE

フィールド構文：-F key=value は型を自動変換し、-f key=value は文字列として保持します。

完全なリファレンス：すべてのオプション、一般的なエンドポイント（XCom、event-logs、backfills）、例については、api-reference.md を参照してください。

関連スキル

スキル	使用する場面...
authoring-dags	ベストプラクティスに従って DAG ファイルを作成または編集するとき
testing-dags	テスト → デバッグ → 修正 → 再テストを繰り返すとき
debugging-dags	根本原因を詳しく分析し、障害を診断するとき
checking-freshness	データが最新か、古くなっているかを確認するとき
tracing-upstream-lineage	データの取得元を調べるとき
tracing-downstream-lineage	影響を分析するとき――何かを変更した場合に何が壊れるか
deploying-airflow	DAG を本番環境（Astro、Docker Compose、Kubernetes）にデプロイするとき
migrating-airflow-2-to-3	DAG を Airflow 2.x から 3.x にアップグレードするとき
managing-astro-local-env	ローカル Airflow を起動、停止、またはトラブルシューティングするとき
setting-up-astro-project	新しい Astro／Airflow プロジェクトを初期化するとき
airflow-state-store	タスク単位のチェックポイント処理、ウォーターマーク、クラッシュセーフなオペレーターを扱うとき（Airflow 3.3+）
airflow-hitl	人による承認や入力のために DAG を一時停止するとき（Airflow 3.1+）