migrating-airflow-2-to-3

作成者: astronomer

Apache Airflow 2.xのDAGをAirflow 3.xにアップグレードするための自動検出とコード移行。Ruffベースの自動修正ルール(AIR30/AIR301/AIR302/AIR31/AIR311/AIR312)を提供し、インポート、オペレーター、フック、コンテキスト変数における破壊的変更を検出・解決します。重要なアーキテクチャの変更に対応:ワーカーはメタデータDBに直接アクセスしなくなり、ORMセッションクエリの代わりにAirflow PythonクライアントまたはREST APIを使用します。Ruffが自動修正できない問題に対する手動移行チェックリストを含む:cron...

npx skills add https://github.com/astronomer/agents --skill migrating-airflow-2-to-3

Airflow 2 to 3 Migration

This skill helps migrate Airflow 2.x DAG code to Airflow 3.x, focusing on code changes (imports, operators, hooks, context, API usage).

Important: Before migrating to Airflow 3, strongly recommend upgrading to Airflow 2.11 first, then to at least Airflow 3.0.11 (ideally directly to 3.1). Other upgrade paths would make rollbacks impossible. See: https://www.astronomer.io/docs/astro/airflow3/upgrade-af3#upgrade-your-airflow-2-deployment-to-airflow-3. Additionally, early 3.0 versions have many bugs - 3.1 provides a much better experience.

Migration at a Glance

  1. Run Ruff's Airflow migration rules to auto-fix detectable issues (AIR30/AIR301/AIR302/AIR31/AIR311/AIR312).
    • ruff check --preview --select AIR --fix --unsafe-fixes .
  2. Scan for remaining issues using the manual search checklist in reference/migration-checklist.md.
    • Focus on: direct metadata DB access, legacy imports, scheduling/context keys, XCom pickling, datasets-to-assets, REST API/auth, plugins, and file paths.
    • Hard behavior/config gotchas to explicitly review:
      • Cron scheduling semantics: consider AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVAL=True if you need Airflow 2-style cron data intervals.
      • .airflowignore syntax changed from regexp to glob; set AIRFLOW__CORE__DAG_IGNORE_FILE_SYNTAX=regexp if you must keep regexp behavior.
      • OAuth callback URLs add an /auth/ prefix (e.g. /auth/oauth-authorized/google).
      • Shared utility imports: Bare imports like import common from dags/common/ no longer work on Astro. Use fully qualified imports: import dags.common.
  3. Plan changes per file and issue type:
    • Fix imports - update operators/hooks/providers - refactor metadata access to using the Airflow client instead of direct access - fix use of outdated context variables - fix scheduling logic.
  4. Implement changes incrementally, re-running Ruff and code searches after each major change.
  5. Explain changes to the user and caution them to test any updated logic such as refactored metadata, scheduling logic and use of the Airflow context.

Architecture & Metadata DB Access

Airflow 3 changes how components talk to the metadata database:

  • Workers no longer connect directly to the metadata DB.
  • Task code runs via the Task Execution API exposed by the API server.
  • The DAG processor runs as an independent process separate from the scheduler.
  • The Triggerer uses the task execution mechanism via an in-process API server.

Trigger implementation gotcha: If a trigger calls hooks synchronously inside the asyncio event loop, it may fail or block. Prefer calling hooks via sync_to_async(...) (or otherwise ensure hook calls are async-safe).

Key code impact: Task code can still import ORM sessions/models, but any attempt to use them to talk to the metadata DB will fail with:

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

Patterns to search for

When scanning DAGs, custom operators, and @task functions, look for:

  • Session helpers: provide_session, create_session, @provide_session
  • Sessions from settings: from airflow.settings import Session
  • Engine access: from airflow.settings import engine
  • ORM usage with models: session.query(DagModel)..., session.query(DagRun)...

Replacement: Airflow Python client

Preferred for rich metadata access patterns. Add to requirements.txt:

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

Example usage:

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))

Replacement: Direct REST API calls

For simple cases, call the REST API directly using requests:

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 Migration Rules

Use Ruff's Airflow rules to detect and fix many breaking changes automatically.

  • AIR30 / AIR301 / AIR302: Removed code and imports in Airflow 3 - must be fixed.
  • AIR31 / AIR311 / AIR312: Deprecated code and imports - still work but will be removed in future versions; should be fixed.

Commands to run (via uv) against the project root:

# 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 Files

For detailed code examples and migration patterns, see:


Quick Reference Tables

Key Import Changes

Airflow 2.xAirflow 3
airflow.operators.dummy_operator.DummyOperatorairflow.providers.standard.operators.empty.EmptyOperator
airflow.operators.bash.BashOperatorairflow.providers.standard.operators.bash.BashOperator
airflow.operators.python.PythonOperatorairflow.providers.standard.operators.python.PythonOperator
airflow.decorators.dagairflow.sdk.dag
airflow.decorators.taskairflow.sdk.task
airflow.datasets.Datasetairflow.sdk.Asset

Context Key Changes

Removed KeyReplacement
execution_datecontext["dag_run"].logical_date
tomorrow_ds / yesterday_dsUse ds with date math: macros.ds_add(ds, 1) / macros.ds_add(ds, -1)
prev_ds / next_dsprev_start_date_success or timetable API
triggering_dataset_eventstriggering_asset_events
templates_dictcontext["params"]

Asset-triggered runs: logical_date may be None; use context["dag_run"].logical_date defensively.

Cannot trigger with future logical_date: Use logical_date=None and rely on run_id instead.

Cron note: for scheduled runs using cron, logical_date semantics differ under CronTriggerTimetable (aligning logical_date with run_after). If you need Airflow 2-style cron data intervals, consider AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVAL=True.

Default Behavior Changes

SettingAirflow 2 DefaultAirflow 3 Default
scheduletimedelta(days=1)None
catchupTrueFalse

Callback Behavior Changes

  • on_success_callback no longer runs on skip; use on_skipped_callback if needed.
  • @teardown with TriggerRule.ALWAYS not allowed; teardowns now execute even if DAG run terminated early.

Resources


Related Skills

  • testing-dags: For testing DAGs after migration
  • debugging-dags: For troubleshooting migration issues
  • deploying-airflow: For deploying migrated DAGs to production

astronomerのその他のスキル

airflow-state-store
astronomer
Persists task and asset state across retries and DAG runs using Airflow 3.3's AIP-103 key/value stores (`task_state_store`, `asset_state_store`) and the…
creating-openlineage-extractors
astronomer
カスタムOpenLineage抽出器。未対応のAirflowオペレーターや複雑な系列シナリオ向け。2つのアプローチ:所有するオペレーターに直接OpenLineageメソッドを追加する方法(推奨)、または変更できないサードパーティ製オペレーター用にカスタム抽出器を作成する方法。抽出器は3つの時点でオペレーターの実行をインターセプトします:静的な系列のための実行前、実行時に決定される出力のための成功後、およびオプションで部分的な系列のための失敗後。抽出器はairflow.cfgまたは環境変数経由で登録...
debugging-dags
astronomer
失敗したAirflow DAGに対する体系的な根本原因分析と修正、構造化された調査ワークフローを提供。4段階の診断プロセス(障害の特定、エラー詳細の抽出、コンテキスト情報の収集、実行可能な修正手順の提示)をガイド。障害を4つのタイプ(データ、コード、インフラストラクチャ、依存関係)に分類し、調査を集中させ適切な修正を提案。ログ取得、実行比較、タスククリア、DAG...のための即時使用可能なCLIコマンドを提供。
delegating-to-otto
astronomer
Drives Astronomer's Otto agent (`astro otto`) as a delegated sub-agent for Airflow, dbt, and data-engineering work. Use when the user explicitly asks to "use…
deploying-airflow
astronomer
AirflowのDAGやプロジェクトをデプロイします。ユーザーがコードのデプロイ、DAGのプッシュ、CI/CDの設定、本番環境へのデプロイ、またはデプロイ戦略について質問した場合に使用します…
deploying-go-sdk-bundles
astronomer
コンパイルされたAirflow Go SDKバンドルをビルド、パッケージ化、デプロイし、ExecutableCoordinatorが実行できるようにします。ユーザーがGoタスクバンドルをコンパイルしたい場合、または…と尋ねた場合に使用します。
testing-dags
astronomer
Airflow DAGに対する反復的なテスト・デバッグ・修正サイクルと、包括的な障害診断を提供します。af runs trigger-wait <dag_id> でDAGを実行し完了を待機します。事前チェックは不要です。失敗時はaf runs diagnoseで包括的な障害サマリーを取得し、af tasks logsで特定タスクのエラー詳細を確認できます。カスタム設定、タイムアウト、リトライ試行に対応。成功、失敗、タイムアウトの各シナリオを明確な応答解釈で処理します。迅速な検証が可能です...
tracing-downstream-lineage
astronomer
テーブルやDAGを変更する前に、下流のデータ系列を追跡して変更影響を評価します。ソースコード検索、ビュー依存関係、BIツール接続を通じて、対象テーブルまたはDAGの直接的な消費者を特定します。テーブルからダッシュボード、MLモデルに至るまで、すべての下流影響をマッピングする完全な依存関係ツリーを構築します。依存関係を重要度(クリティカル、高、中、低)で分類し、ステークホルダーへの連絡とテストの優先順位付けを行います。リスク評価と影響を受けるものを含む影響レポートを生成します。