Claude Media
Claude CodeのAstronomerプラグインでAirflow DAGを書いて直す
Claude CodeMCPClaude連携チュートリアル

Claude CodeのAstronomerプラグインでAirflow DAGを書いて直す

Astronomerのデータエンジニアリングプラグインを入れ、authoring-dagsでDAGを書き、debugging-dagsで失敗を直し、Airflow 3への移行とCosmosまで進める手順です。

読了目安 約14分

Astronomerが配布するデータエンジニアリングプラグインは、Airflow用のMCPサーバーと30以上のスキルを1回の導入でまとめて入れられます。DAGの新規作成、失敗の原因調査、Airflow 2から3への移行、dbtプロジェクトのDAG化までを、スキルが決まった手順で進めます。ここでは入れ方、DAGを書くスキルと直すスキル、移行とCosmosの扱い、安全に任せるための設定を順に追います。

プラグインの入れ方と最初の確認

導入は2コマンドです。マーケットプレイスの表示名は「Data Engineering」ですが、GitHubのREADMEが示すインストール名はastronomer-data@astronomerです。

claude plugin marketplace add astronomer/agents
claude plugin install astronomer-data@astronomer

以前の名前data@astronomerで入れた場合は、先にアンインストールしてから入れ直します。スキルが一覧に出ないときも同じ手順で再導入します。

claude plugin uninstall data@astronomer \
  && claude plugin marketplace update \
  && claude plugin install astronomer-data@astronomer

プラグインにはAirflow MCPサーバーが含まれ、PyPIからuvxで起動します。Airflowのプロジェクトディレクトリ(airflow.cfgかdags/がある場所)でClaude Codeを起動すると、接続先を自動で見つけます。リモートのAirflowを使うなら、次の環境変数を設定します。

環境変数役割
AIRFLOW_API_URL役割Airflow webserverのURL
AIRFLOW_USERNAME役割ユーザー名
AIRFLOW_PASSWORD役割パスワード
AIRFLOW_AUTH_TOKEN役割Bearerトークン(ユーザー名とパスワードの代わり)

スキルは自動で呼ばれるほか、/astronomer-data:<スキル名>で直接呼べます。MCPサーバーだけを単体で使う方法はAirflow MCPサーバーの記事にあります。

authoring-dagsはDAGを6段階で書く

「S3からSnowflakeへ毎時取り込むDAGを作って」のような依頼で動くのがauthoring-dagsです。コードを書き始める前に環境を調べ、計画の承認を取ってから実装します。

手順

authoring-dagsの流れ

  1. 1

    Discover

    既存のDAGと慣習をコードから探し、afコマンドで接続・変数・インストール済みプロバイダーを確かめます。

  2. 2

    Plan

    DAGの構成、使うオペレーター、必要な接続と変数、追加パッケージを提案し、実装前にユーザーの承認を取ります。

  3. 3

    Implement

    DAGファイルを作り、必要ならrequirements.txtを更新します。

  4. 4

    Validate

    インポートエラー、DAGの存在、警告をafで順に確かめます。

  5. 5

    Test

    ユーザーの同意を得てからDAGを起動し、結果とログを見ます。

  6. 6

    Iterate

    問題があればコードを直し、検証からやり直します。

Discoverで使うafコマンドは、問いごとに決まっています。

知りたいことコマンド
どの外部システムの接続があるかコマンドaf config connections
どんな設定値(変数)があるかコマンドaf config variables
どのオペレーターパッケージが入っているかコマンドaf config providers
Airflowのバージョンと制約コマンドaf config version
既存DAGと命名規則コマンドaf dags list
並列実行用のプールコマンドaf config pools

Validateでは、af dags errorsでインポートエラーを先に見て、af dags get <dag_id>でスケジュールとタグとpause状態を確かめ、af dags warningsで非推奨の警告を拾います。af dags explore <dag_id>なら、メタデータ・タスク・依存関係・ソースコードを1回で返します。Astronomer製のAstroを使っているなら、astro dev parseでAirflowを起動せずにパース確認もできます。

dags testやインポートエラーの切り分けを手元で追う手順は、DAG検証の記事が詳しく扱っています。プラグインを使う場合、その検証をClaudeがafコマンドで自走するのが違いです。

Testの段階で必ず同意を取る点は、スキルの手順に書かれています。実行にはaf runs trigger-wait <dag_id> --timeout 300を使います。さらに、このスキルは停止時のフックを持ちます。セッションを止めると「testing-dagsスキルでDAGをテストするのを忘れずに」というメッセージが出ます。

debugging-dagsは失敗を4種類に分けて原因を探す

「昨夜のETLがなぜ失敗した?」「DAGが壊れたので直して」に応えるのがdebugging-dagsです。ログを読むだけの単純な質問は、メインのairflowスキルが直接答えます。深掘りが要るときにこちらが選ばれます。

手順は4ステップです。

  1. 失敗の特定: 実行IDがあればaf runs diagnose <dag_id> <dag_run_id>、なければaf dags statsやaf healthで直近の失敗を探します。
  2. エラーの詳細: af tasks logs <dag_id> <dag_run_id> <task_id>でログを読み、Airflow定型文の奥にある実際の例外を探します。
  3. 背景の確認: 直近のデプロイ、データ量の急増、上流の状態、過去の同種の失敗、発生時刻を調べます。
  4. 出力: 根本原因、影響範囲、即時の修正、再発防止、すぐ使えるコマンドの順にまとめます。

失敗は、データの問題・コードの問題・インフラの問題・依存の問題の4種類に分類されます。

コードを触っていないのに壊れたとき

このスキルの特徴は、パッケージのバージョン変動を独立した調べ方にしている点です。コードが変わっていないのに失敗するケースの典型が、依存パッケージの更新だからです。

  • 直近2つのワーカーイメージでpip freezeの差分を取ります。Pythonのバージョン差も併せて比べます。
  • @task.virtualenvやKubernetesPodOperatorなどは、ワーカーイメージと別に環境を作ります。このためイメージの差分では見えず、requirementsやimageの指定を直接読みます。
  • 上限のない指定(pandas>=2.0.0)や:latestのイメージが疑わしい、という優先順位です。
  • イメージ履歴で決着しなければ、PyPIなどのインデックスで該当パッケージのリリース日時を調べます。初めて失敗した実行より前、最後に成功した実行より後に出たリリースが原因候補です。

修正の例は、pandas>=2.0.0,<3.0.0のように上限を付ける、ロックファイルを使う、イメージをSHAで固定する、といったものです。

直したあとの再実行用に、スキルは次のコマンドを案内します。

af runs clear <dag_id> <run_id>               # 実行全体をやり直す
af tasks clear <dag_id> <run_id> <task_ids> -D # 失敗したタスクだけやり直す
af runs delete <dag_id> <run_id>               # 詰まった実行を削除する

migrating-airflow-2-to-3はRuffから始める

Airflow 3への移行にはmigrating-airflow-2-to-3が使えます。DAGコードの変更(インポート、オペレーター、フック、コンテキスト、APIの使い方)が対象です。

最初に、RuffのAIRルールで自動修正できる箇所を直します。

ruff check --preview --select AIR --fix --unsafe-fixes .

AIR30・AIR301・AIR302はAirflow 3で削除された記述で、必ず直す対象です。AIR31・AIR311・AIR312は非推奨で、今は動いても将来消えます。Ruffで拾えないものは、スキル付属のチェックリストで手作業の検索になります。

スキルが手順の冒頭で強調する注意点は、アップグレードの経路です。まずAirflow 2.11に上げ、次に3.0.11以上(できれば3.1へ直接)へ進めます。ほかの経路だとロールバックできなくなります。

移行で特に踏みやすい変更は次のとおりです。

項目Airflow 2Airflow 3
scheduleの既定値Airflow 2timedelta(days=1)Airflow 3None
catchupの既定値Airflow 2TrueAirflow 3False
execution_dateAirflow 2コンテキストのキーAirflow 3context["dag_run"].logical_date
DatasetAirflow 2airflow.datasets.DatasetAirflow 3airflow.sdk.Asset
@dag / @taskAirflow 2airflow.decoratorsAirflow 3airflow.sdk

ワーカーからメタデータDBへ直接つなぐ書き方は、Airflow 3では動きません。provide_sessionやORMのクエリを使っていると、「Direct database access via the ORM is not allowed in Airflow 3.x」というエラーになります。スキルは代わりにAirflowのPythonクライアントかREST APIの利用を案内します。

cosmos-dbt-coreはdbtプロジェクトをDAGにする

dbt CoreのプロジェクトをAirflowのDAGやTaskGroupにするのがcosmos-dbt-coreです。Astronomer Cosmosを使います。このスキルは実装の前に6点の確認を求めます。

  • dbtのエンジンがCoreか(Fusionなら別スキルのcosmos-dbt-fusion)
  • ウェアハウスの種類
  • Airflowのバージョン
  • 実行環境(Airflow環境か、仮想環境か、コンテナか)
  • DbtDagかDbtTaskGroupか、個別のオペレーターか
  • manifestが使えるか

スキルはCosmos 1.11以降とAirflow 3.xを前提に書かれています。Airflow 2.xのインポート差は付録で扱います。cosmos-dbt-fusionが対応するのはSnowflakeとDatabricksだけです。

Claudeに任せる範囲を設定で決める

スキルの手順には「起動前に同意を取る」とありますが、手順は指示であって強制ではありません。Claude Code側の権限で、読むだけのafコマンドは通し、実行を伴うものは確認を挟む構成にできます。次はその一例です(プラグインの付属物ではなく、権限設定の書き方に沿った筆者の例示です)。

{
  "permissions": {
    "allow": [
      "Bash(af dags errors)",
      "Bash(af dags list)",
      "Bash(af dags get:*)",
      "Bash(af dags warnings)",
      "Bash(af config:*)",
      "Bash(af tasks logs:*)"
    ],
    "ask": [
      "Bash(af runs trigger-wait:*)",
      "Bash(af runs clear:*)",
      "Bash(af tasks clear:*)"
    ],
    "deny": ["Bash(af runs delete:*)"]
  }
}

DAGの命名規則やチームの決まりごとは、CLAUDE.mdに書いておくとDiscoverの段階で拾われます。例えば次のような断片です。

# Airflow規約
- DAG IDは `<チーム>_<対象>_<頻度>` にする
- 接続は `af config connections` に存在するものだけを使う
- スキーマ変更を伴うDAGは、起動前に必ず確認を取る

つまずきやすい点

落とし穴

Airflowが検出されない

自動検出は、airflow.cfgかdags/のあるディレクトリでClaude Codeを起動したときに働きます。別の場所から起動すると見つかりません。

落とし穴

ウェアハウス接続のエラー

データ分析系のスキルは~/.astro/agents/warehouse.ymlと~/.astro/agents/.envを読みます。認証情報を.envに置き、warehouse.ymlから${変数名}で参照します。SnowflakeのアカウントにはアカウントIDの識別子(orgname-accountname形式など)を指定します。

  • afコマンドは、uv tool install astro-airflow-mcpで入れるか、uvx --from astro-airflow-mcp af --helpで直接実行します。
  • afは匿名の利用統計を取ります。収集されるのはコマンド名だけで、引数や値は含まれません。止めるにはaf telemetry disableです。
  • ヒューマンインザループのスキルairflow-hitlは、Airflow 3.1以降が対象です。
  • Airflow 2.xやオープンソース版でもMCPサーバーは動きます。AstroのセットアップスキルはAstro専用ですが、DAGを書くスキルは環境を選びません。

まとめ

DAGを書く・直す・移行する作業を、同じafコマンド群の上で回せる点がこのプラグインの強みです。まずはAirflow 3対応の検証用プロジェクトで、afの読み取り系だけを許可してauthoring-dagsを試すと、挙動を安全に確かめられます。

この記事を共有:XはてブLinkedIn

関連する記事

Claude Code をもっと見る →