Airflow 2 to 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 pickle、データセット→アセット、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 プロセッサはスケジューラから独立したプロセスとして実行されます。
• Triggerer はインプロセス API サーバー経由でタスク実行メカニズムを使用します。

Trigger 実装の注意点: トリガーが 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 経由):

# 検出可能なすべての Airflow 問題を自動修正(安全 + 不安全)
ruff check --preview --select AIR --fix --unsafe-fixes .

# 修正せずに残りの Airflow 問題をチェック
ruff check --preview --select AIR .

---

リファレンスファイル

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

• reference/config-changes.md - airflow.cfg セクションの移動、名前変更、削除
• reference/migration-patterns.md - インポート、スケジューリング、XCom、アセット、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 またはタイムテーブル API
triggering_dataset_events	triggering_asset_events
templates_dict	context["params"]

アセットトリガーされた実行: 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 を使用してください。
• TriggerRule.ALWAYS を持つ @teardown は許可されていません。テアダウンは DAG 実行が早期に終了した場合でも実行されます。

---

リソース

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

---

関連スキル

• testing-dags: 移行後の DAG テスト用
• debugging-dags: 移行問題のトラブルシューティング用
• deploying-airflow: 移行された DAG の本番環境へのデプロイ用