🚀

Claude Codeで80個のDAGをAirflow 3.0に移行してみた

に公開
Airflow
tech

この記事は一休.com Advent Calendar 2025の9日目の記事です。
https://qiita.com/advent-calendar/2025/ikyu

80個程度のDAGを持っているApache AirflowのメジャーバージョンアップをClaude Codeに任せたらどうなるか？という検証をしてみようと思います。
手元には古いバージョンのairflowがあり、実際にLLM任せで人間が介入せずバージョン更新ができるか検証してみた記事です。

airflowについて

https://airflow.apache.org

  • 複雑なバッチ処理を依存関係つきでオーケストレーションする仕組み
  • データパイプラインの可視化・監視・運用を一元化できる
  • スケジューラが定期実行や再実行を自動で管理する

OSSなので自前インフラで構築できますが、マネージドサービスも存在します。

  • AWSだとAmazon Managed Workflows for Apache Airflow (MWAA)
  • GCPだとCloud Composer

そのapache airflowがいつの間にかバージョン3に上がっていました。

https://airflow.apache.org/docs/apache-airflow/stable/installation/upgrading_to_airflow3.html

airflow 3.x系の主な破壊的変更の内容

  • SubDagの廃止によるTaskGroupsへの置き換えと、Assetsの使用推奨
  • airflow module関連のImport pathの変更

特にSubDagの廃止などは2.xで非推奨になりつつも置き換えが面倒で使い続けていた方も多そうです。
1系、2系どちらを使っていても破壊的変更が多いので、どうしても大量のタスクを置き換えるのが億劫になっている私のような人は多いのではないでしょうか？

具体的な変換例

Before（旧バージョン）

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.subdag_operator import SubDagOperator

subdag_task = SubDagOperator(
    task_id='my_subdag',
    subdag=MyClass.get_subdag(dag.dag_id, default_args),
    dag=dag
)

After（Airflow 3.0）

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import TaskGroup

my_task_group = MyClass.create_task_group(dag, default_args)

このようなパスの変換、記法をほぼ人の手を介さずに自動で行うようにしたいと思います。

マイグレーションにおける課題

  • 手元には80個以上のSubDagファイルがある
  • 人手でやると膨大な時間がかかる
  • しかも単純作業の繰り返し（パターン変換が結構ある）
  • マイグレーションかつ最低限の動作確認もLLMに任せられるのか？

ゴールの定義

  • Airflow 3.0 で最低限の動作確認ができる事
  • できるだけ人間の作業を減らす
  • 起動時にエラーが出ない
  • 人間が細かいDAG内の検証を行えるためのスタートラインに持っていく

プロンプト（指示書）の設計

私がやりたいことは、「LLMに指示を出して完了したらairflowが3.0系で動いていること」
これを実現するための指示書の設計はとても大事なので、指示を明確化します。

指示書の内容

やって欲しいこと
- 既存の振る舞いを保ったまま、レガシーコードを Airflow 3.0 互換へ刷新する。
- まずは import / 非推奨 API / 文法を中心に「機械的に安全に」直す。
- 変更は最小限・可読性高く。副作用を入れない。

制約（やってはいけないこと）を明確に
- 動かない魔改造はNG。逐次的・小さなパッチにまとめる。
- 互換のないAPIは後方互換レイヤ（TaskGroup, TaskFlow API等）で置き換え。
- 変数名・関数名は基本維持
- 致命的、または大幅な変更を余儀なくされた場合は許可を求めること
- dagごとにマイグレーションを完了したら、{dag名}_migration概要..mdとして作業概要を記して完了としてください

検証手順
- ruff check dags/ {全てのディレクトリを記述} --select AIR3 コマンドを使って全てのdagがlintを通るようにする
- 次に、make restart コマンドでコンテナを再起動し、エラーが出ていないか確認する

LLMが自分で検証できるように、ruff checkやdockerの起動コマンドやlogの見方などをまとめたmakefileを作り、使い方を提示することでLLMに検証手順を伝えるようにしました。

実行する

Phase 1: 小規模DAGで試行

  • まずは検証のために小さいDAGで試行
  • 指示書の調整

Phase 2: 中規模DAGへ拡大

  • パターンを確立する

Phase 3: 本丸の大規模DAGの移行

  • 80個のSubDagを持つ巨大DAG
  • 約4時間で完了

やってみて思ったこと

うまくいったこと

  • ruff check, docker logのチェック、単体テストなど、「検証可能な作業」を自動化出来るケースは相性がいい
  • 検証手段を提示することで、LLMが自分で「正しいかどうか」を判断できた

うまくいかなかったこと

ruffでは検出できないエラーがあった

LLMが変換した後、ruff checkは通るのにDockerで起動するとエラーが95件出ていました。
原因は with TaskGroup の直後に空行があると Python構文エラーになるという単純な問題で、構文チェックは通っても起動エラーになるケースは何度かコンテナログをチェックしてもらい、解消できました。

# これがエラー（ruffは検出しない）
with TaskGroup(...) as tg:
                      # ← この空行がダメ
    task = PythonOperator(...)

# 正しい
with TaskGroup(...) as tg:
    task = PythonOperator(...)

終わりに

LLM任せでバージョンアップして見た結果は、概ね期待通りにできたと思います。
作業を任せたことで1人で長い時間かかるコードの書き換えと検証作業も数時間という短い時間で終えることができました。
これからも積極的に活用したいですね。

Discussion