migrating-airflow-2-to-3

Airflow 2 から 3 への移行

このスキルは、インポート、オペレーター、フック、コンテキスト、API の使用方法などのコード変更に重点を置き、Airflow 2.x の DAG コードを Airflow 3.x に移行するのに役立ちます。

重要：Airflow 3 に移行する前に、まず Airflow 2.11 にアップグレードし、その後、少なくとも Airflow 3.0.11（できれば 3.1）にアップグレードすることを強くおすすめします。

移行の概要

1. Ruff の Airflow 移行ルールを実行します：
   • ruff check --preview --select AIR --fix --unsafe-fixes .
2. 手動チェックリストを使って、残っている問題を確認します：
   • メタデータ DB への直接アクセス
   • レガシーなインポート
   • スケジュール設定とコンテキストの変更
   • XCom の pickle 化
   • データセットからアセットへの移行
   • REST API と認証
   • プラグインとファイルパス
3. ファイルと問題の種類ごとに変更を計画します
4. 変更を段階的に実施し、Ruff を再実行します
5. 変更内容を説明し、テストを推奨します

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

Airflow 3 では、コンポーネントがメタデータデータベースと通信する方法が変わります：

• ワーカーはメタデータ DB に直接接続しなくなります
• タスクコードは Task Execution API 経由で実行されます
• DAG プロセッサーはスケジューラーから分離されています

ORM に直接アクセスすると、次のエラーが発生します：

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

検索するパターン

• provide_session、create_session、@provide_session
• from airflow.settings import Session
• from airflow.settings import engine
• モデルを使った ORM の使用：session.query(DagModel)...

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

requirements.txt に追加します：

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

例：

import os
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")

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)

Ruff の Airflow 移行ルール

• AIR30 / AIR301 / AIR302：削除されたコードとインポート
• AIR31 / AIR311 / AIR312：非推奨のコードとインポート

コマンド：

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

主なインポートの変更

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	macros.ds_add を使用します
triggering_dataset_events	triggering_asset_events
templates_dict	context["params"]

デフォルト動作の変更

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

リソース

• https://www.astronomer.io/docs/astro/airflow3/upgrade-af3
• https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html