migrating-airflow-2-to-3

Airflow 2 から 3 への移行

このスキルは、Airflow 2.x の DAG コードをAirflow 3.xへ移行する際に、コードの変更（インポート、オペレーター、フック、コンテキスト、API の使用）に重点を置いて支援します。

重要：Airflow 3 に移行する前に、まず Airflow 2.11 にアップグレードし、その後少なくとも Airflow 3.0.11（できれば直接 3.1）にアップグレードすることを強く推奨します。これ以外のアップグレード経路では、ロールバックが不可能になります。詳細は https://www.astronomer.io/docs/astro/airflow3/upgrade-af3#upgrade-your-airflow-2-deployment-to-airflow-3 を参照してください。また、初期の 3.0 バージョンには多くのバグがあり、3.1 では大幅に使いやすくなっています。

移行の概要

1. Ruff の Airflow 移行ルールを実行し、自動修正可能な問題を修正します（AIR30/AIR301/AIR302/AIR31/AIR311/AIR312）。
   • ruff check --preview --select AIR --fix --unsafe-fixes .
2. reference/migration-checklist.md の手動検索チェックリストを使って、残りの問題を確認します。
   • 次の項目に重点を置きます：メタデータ DB への直接アクセス、古いインポート、スケジューリング／コンテキストのキー、XCom のピクル化、データセットからアセットへの移行、REST API／認証、プラグイン、ファイルパス。
   • 動作や設定に関する重要な注意点は、必ず確認してください：
     • cron スケジューリングのセマンティクス：Airflow 2 形式の cron データ間隔が必要な場合は、AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVAL=True を検討してください。
     • .airflowignore の構文が正規表現から glob に変更されました。正規表現の動作を維持する必要がある場合は、AIRFLOW__CORE__DAG_IGNORE_FILE_SYNTAX=regexp を設定してください。
     • OAuth コールバック URL に /auth/ プレフィックスが追加されます（例：/auth/oauth-authorized/google）。
     • 共有ユーティリティのインポート：dags/common/ からの import common のようなプレフィックスのないインポートは、Astro では機能しなくなりました。完全修飾インポート（import dags.common）を使用してください。
3. ファイルと問題の種類ごとに変更を計画します：
   • インポートを修正する - オペレーター／フック／プロバイダーを更新する - 直接アクセスの代わりに Airflow クライアントを使うようメタデータアクセスをリファクタリングする - 古いコンテキスト変数の使用を修正する - スケジューリングロジックを修正する。
4. 段階的に変更を実装し、大きな変更のたびに Ruff とコード検索を再実行します。
5. 変更内容をユーザーに説明し、リファクタリングしたメタデータ、スケジューリングロジック、Airflow コンテキストの使用など、更新したロジックをテストするよう注意を促します。

---

アーキテクチャとメタデータ DB へのアクセス

Airflow 3 では、コンポーネントがメタデータデータベースと通信する方法が変更されています：

• ワーカーはメタデータ DB に直接接続しなくなりました。
• タスクコードは、API サーバーが公開するタスク実行 APIを介して実行されます。
• DAG プロセッサーは、スケジューラーとは別の独立したプロセスとして実行されます。
• トリガーは、プロセス内 API サーバーを介してタスク実行の仕組みを使用します。

トリガー実装に関する注意点：トリガーが asyncio イベントループ内でフックを同期的に呼び出すと、失敗したりブロックしたりする場合があります。sync_to_async(...) を介してフックを呼び出すか、フックの呼び出しが非同期処理に対応していることを確認してください。

コードへの主な影響：タスクコードから ORM セッション／モデルをインポートすることは引き続き可能ですが、メタデータ DB へのアクセスに使用しようとすると、すべて失敗します：

RuntimeError: Direct database access via the ORM is not allowed in Airflow 3.x

検索するパターン

DAG、カスタムオペレーター、@task 関数を確認する際は、次を探してください：

• セッションヘルパー：provide_session、create_session、@provide_session
• 設定からのセッション：from airflow.settings import Session
• エンジンへのアクセス：from airflow.settings import engine
• モデルを使用した ORM：session.query(DagModel)...、session.query(DagRun)...

置き換え：Airflow Python クライアント

高度なメタデータアクセスパターンにはこちらが推奨されます。requirements.txt に追加します：

apache-airflow-client==<your-airflow-runtime-version>

使用例：

import os
from airflow.sdk import BaseOperator
import airflow_client.client
from airflow_client.client.api.dag_api import DAGApi

_HOST = os.getenv("AIRFLOW__API__BASE_URL", "https://<your-org>.astronomer.run/<deployment>/")
_TOKEN = os.getenv("DEPLOYMENT_API_TOKEN")

class ListDagsOperator(BaseOperator):
    def execute(self, context):
        config = airflow_client.client.Configuration(host=_HOST, access_token=_TOKEN)
        with airflow_client.client.ApiClient(config) as api_client:
            dag_api = DAGApi(api_client)
            dags = dag_api.get_dags(limit=10)
            self.log.info("Found %d DAGs", len(dags.dags))

置き換え：REST API の直接呼び出し

単純なケースでは、requests を使って REST API を直接呼び出します：

from airflow.sdk import task
import os
import requests

_HOST = os.getenv("AIRFLOW__API__BASE_URL", "https://<your-org>.astronomer.run/<deployment>/")
_TOKEN = os.getenv("DEPLOYMENT_API_TOKEN")

@task
def list_dags_via_api() -> None:
    response = requests.get(
        f"{_HOST}/api/v2/dags",
        headers={"Accept": "application/json", "Authorization": f"Bearer {_TOKEN}"},
        params={"limit": 10}
    )
    response.raise_for_status()
    print(response.json())

---

Ruff Airflow 移行ルール

Ruff の Airflow ルールを使うと、多くの破壊的変更を自動的に検出して修正できます。

• AIR30 / AIR301 / AIR302：Airflow 3 で削除されたコードとインポートです。必ず修正してください。
• AIR31 / AIR311 / AIR312：非推奨のコードとインポートです。現在も動作しますが、今後のバージョンで削除されるため、修正することを推奨します。

プロジェクトのルートで uv を使って実行するコマンド：

# Auto-fix all detectable Airflow issues (safe + unsafe)
ruff check --preview --select AIR --fix --unsafe-fixes .

# Check remaining Airflow issues without fixing
ruff check --preview --select AIR .

---

参考ファイル

詳しいコード例と移行パターンについては、以下を参照してください。

• reference/config-changes.md - airflow.cfg のセクションの移動、名前変更、削除
• reference/migration-patterns.md - インポート、スケジューリング、XCom、Assets、DAG バンドル、実行時の動作変更に関するコード例
• reference/removed-methods.md - 削除されたモデルメソッドと、SDK/API への移行方法
• reference/migration-checklist.md - Ruff で検出できない問題の検索パターンと修正方法

---

クイックリファレンス表

主なインポートの変更

Airflow 2.x	Airflow 3
airflow.operators.dummy_operator.DummyOperator	airflow.providers.standard.operators.empty.EmptyOperator
airflow.operators.bash.BashOperator	airflow.providers.standard.operators.bash.BashOperator
airflow.operators.python.PythonOperator	airflow.providers.standard.operators.python.PythonOperator
airflow.decorators.dag	airflow.sdk.dag
airflow.decorators.task	airflow.sdk.task
airflow.datasets.Dataset	airflow.sdk.Asset

コンテキストキーの変更

削除されたキー	置き換え先
execution_date	context["dag_run"].logical_date
tomorrow_ds / yesterday_ds	日付計算で ds を使用します：macros.ds_add(ds, 1) / macros.ds_add(ds, -1)
prev_ds / next_ds	prev_start_date_success または timetable API
triggering_dataset_events	triggering_asset_events
templates_dict	context["params"]

Asset によってトリガーされる実行：logical_date が None の場合があります。防御的に context["dag_run"].logical_date を使用してください。

未来の logical_date ではトリガーできません：logical_date=None を使用し、代わりに run_id を利用してください。

Cron に関する注意：cron を使用するスケジュール実行では、CronTriggerTimetable において logical_date の意味が異なります（logical_date が run_after と一致します）。Airflow 2 形式の cron データ間隔が必要な場合は、AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVAL=True を検討してください。

デフォルト動作の変更

設定	Airflow 2 のデフォルト	Airflow 3 のデフォルト
schedule	timedelta(days=1)	None
catchup	True	False

コールバック動作の変更

• on_success_callback はスキップ時に実行されなくなりました。必要に応じて on_skipped_callback を使用してください。
• @teardown で TriggerRule.ALWAYS は使用できません。teardown は DAG 実行が早期終了した場合でも実行されるようになりました。

---

リソース

• Astronomer Airflow 3 アップグレードガイド
• Airflow 3 リリースノート
• Ruff Airflow ルール

---

関連スキル

• testing-dags：移行後の DAG のテスト
• debugging-dags：移行に関する問題のトラブルシューティング
• deploying-airflow：移行した DAG の本番環境へのデプロイ