Airflow Operations

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

Astro CLI

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

# 新しいプロジェクトを初期化
astro dev init

# ローカル Airflow を起動 (webserver は http://localhost:8080)
astro dev start

# DAG をパースしてエラーを素早く検出 (Airflow の起動は不要)
astro dev parse

# DAG に対して pytest を実行
astro dev pytest

# 本番環境にデプロイ
astro deploy            # フルデプロイ (image + DAGs)
astro deploy --dags     # DAG のみデプロイ (高速、image ビルドなし)

詳細情報：

• 新しいプロジェクト？ setting-up-astro-project スキルを参照
• ローカル環境？ managing-astro-local-env スキルを参照
• デプロイ？ deploying-airflow スキルを参照

---

CLI の実行

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

インスタンス設定

複数の Airflow インスタンスを永続設定で管理します：

# 新しいインスタンスを追加
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

# インスタンスをリスト表示して切り替え
af instance list      # すべてのインスタンスをテーブルで表示
af instance use prod  # prod インスタンスに切り替え
af instance current   # 現在のインスタンスを表示
af instance delete old-instance

# インスタンスを自動検出 (実行前に --dry-run でプレビュー)
af instance discover --dry-run        # すべての検出可能なインスタンスをプレビュー
af instance discover                  # すべてのバックエンド (astro, local) から検出
af instance discover astro            # Astro デプロイメントのみ検出
af instance discover astro --all-workspaces  # アクセス可能なすべてのワークスペースを含める
af instance discover local            # 一般的なローカル Airflow ポートをスキャン
af instance discover local --scan     # すべてのポート 1024-65535 をディープスキャン

# 重要: 常に --dry-run で実行してから、ユーザーの同意を得た上で
# discover を実行してください。非 dry-run モードは Astro Cloud で API トークンを作成します。
# これはユーザーの明示的な承認が必要な敏感な操作です。

# インスタンスの出所を表示 (ファイルパス + スコープ)
af instance show prod

# 環境変数でインスタンスを単一のコマンドのみオーバーライド
AIRFLOW_API_URL=https://staging.example.com AIRFLOW_AUTH_TOKEN=$STG af dags list

# または永続的に切り替え
af instance use staging

設定レイアウト (git config システム/グローバル/ローカルをミラー)：

スコープ	ファイル	コミット済み？
グローバル	~/.astro/config.yaml	不適用 (ユーザーごと)
プロジェクト共有	<root>/.astro/config.yaml	はい
プロジェクトローカル	<root>/.astro/config.local.yaml	いいえ (gitignored)

<root> は cwd から上に向かって .astro/ を探して見つかります。プロジェクト内でのデフォルト書き込みルーティング: add/discover → project-shared、use → project-local。--global / --project / --local でオーバーライド可能です。単一ファイルを使用するには AF_CONFIG=<path> を設定してレイアリングをバイパスします。

設定内のトークンは ${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
# またはユーザー名/パスワード:
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>	トリガーして完了を待機
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 レジストリ内のプロバイダーをリスト表示
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 をデプロイ」/ 「本番環境に push」 -> 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 を検証できます：

# DAG をパースしてインポートエラーと構文の問題を検出
astro dev parse

# ユニットテストを実行
astro dev pytest

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

af dags errors     # パース/インポートエラーを確認
af dags warnings   # 廃止予定警告を確認

コードを作成する前にオペレーターシグネチャを検出

airflow.apache.org/registry の Airflow レジストリは、プロバイダークラスとその現在のコンストラクタシグネチャの権威ある情報源です。DAG を作成する際、メモリや古いドキュメントより優先して使用してください—レジストリは live プロバイダーリリースを反映しています。

# すべてのプロバイダーをリスト表示して必要なものを選択
af registry providers | jq '.providers[] | {id, name, version}'

# プロバイダー内のすべてのオペレーター / フック / センサーをリスト表示 (例: standard, amazon, google)
af registry modules standard \
  | jq '.modules[] | {name, type, import_path, docs_url}'

# 特定のクラスの現在のコンストラクタシグネチャを取得
af registry parameters standard \
  | jq '.classes["airflow.providers.standard.operators.hitl.ApprovalOperator"].parameters'

# 部分文字列でモジュールをフィルタリング (概念はわかるがクラス名がわからない場合に便利)
af registry modules standard \
  | jq '.modules[] | select(.import_path | test("hitl"))'

結果はローカルにキャッシュされます: 最新バージョンは 1 時間、ピン留めされたバージョン (変更不可) は 30 日間。任意の modules / parameters / connections 呼び出しに --version X.Y.Z を追加して特定のリリースをターゲットします。

失敗した実行を調査

# 1. 最近の実行をリスト表示して失敗を検出
af runs list --dag-id my_dag

# 2. 特定の実行を診断
af runs diagnose my_dag manual__2024-01-15T10:00:00+00:00

# 3. 失敗したタスクのログを取得 (診断出力から)
af tasks logs my_dag manual__2024-01-15T10:00:00+00:00 extract_data

# 4. 修正後、実行をクリアしてすべてのタスクを再試行
af runs clear my_dag manual__2024-01-15T10:00:00+00:00

朝の健全性チェック

# 1. 全体的なシステム健全性
af health

# 2. 壊れた DAG がないか確認
af dags errors

# 3. プール使用率を確認
af config pools

DAG を理解

# 包括的な概要を取得 (メタデータ + タスク + ソース)
af dags explore my_dag

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

# 一時停止しているか確認
af dags get my_dag

# インポートエラーを確認
af dags errors

# 最近の実行を確認
af runs list --dag-id my_dag

トリガーと監視

# オプション 1: トリガーして待機 (ブロッキング)
af runs trigger-wait my_dag --timeout 1800

# オプション 2: トリガーして後で確認
af runs trigger my_dag
af runs get my_dag <run_id>

出力形式

すべてのコマンドは JSON を出力します (インスタンスコマンドは人間が読める形式のテーブルを使用)：

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

フィルタリングに jq を使用：

# 失敗した実行を検出
af runs list | jq '.dag_runs[] | select(.state == "failed")'

# DAG ID のみを取得
af dags list | jq '.dags[].dag_id'

# 一時停止している DAG を検出
af dags list | jq '[.dags[] | select(.is_paused == true)]'

タスクログオプション

# 特定の再試行の試みに対するログを取得
af tasks logs my_dag run_id task_id --try 2

# マップされたタスクインデックスのログを取得
af tasks logs my_dag run_id task_id --map-index 5

af api での直接 API アクセス

高レベルコマンドで対応されていないエンドポイント (XCom、event-logs、backfills など) に対して af api を使用します。

# 利用可能なエンドポイントを検出
af api ls
af api ls --filter variable

# 基本的な使用方法
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 プロジェクトを初期化