migrating-airflow-2-to-3

Airflow 2 迁移到 Airflow 3

此技能可帮助将 Airflow 2.x DAG 代码迁移到 Airflow 3.x，重点关注代码变更（导入、运算符、Hook、上下文、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 序列化、从 datasets 迁移到 assets、REST API/身份验证、插件和文件路径。
   • 需要明确检查的行为/配置注意事项：
     • Cron 调度语义：如果需要 Airflow 2 风格的 cron 数据间隔，请考虑设置 AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVAL=True。
     • .airflowignore 语法已从 regexp 改为 glob；如果必须保留 regexp 行为，请设置 AIRFLOW__CORE__DAG_IGNORE_FILE_SYNTAX=regexp。
     • OAuth 回调 URL 会添加 /auth/ 前缀（例如 /auth/oauth-authorized/google）。
     • 共享工具导入：在 Astro 上，import common 这类从 dags/common/ 导入的裸导入不再有效。请使用完整限定导入：import dags.common。
3. 按文件和问题类型规划变更：
   • 修复导入 - 更新运算符/Hook/Provider - 将元数据访问重构为使用 Airflow 客户端，而不是直接访问 - 修复过时的上下文变量用法 - 修复调度逻辑。
4. 逐步实施变更，每完成一项主要变更后都重新运行 Ruff 和代码搜索。
5. 向用户说明变更，并提醒他们测试更新后的逻辑，例如重构后的元数据访问、调度逻辑和 Airflow 上下文用法。

---

架构与元数据 DB 访问

Airflow 3 改变了各组件与元数据数据库通信的方式：

• Worker 不再直接连接元数据 DB。
• 任务代码通过 API 服务公开的 任务执行 API运行。
• DAG 处理器作为独立进程运行，与调度器分离。
• 触发器通过进程内 API 服务使用任务执行机制。

触发器实现注意事项：如果触发器在 asyncio 事件循环中同步调用 Hook，可能会失败或造成阻塞。建议通过 sync_to_async(...) 调用 Hook（或以其他方式确保 Hook 调用与异步操作安全兼容）。

对代码的主要影响：任务代码仍可导入 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 bundles 和运行时行为变更的代码示例
• 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 搭配使用；现在，即使 DAG 运行提前终止，也会执行 teardown。

---

资源

• Astronomer Airflow 3 升级指南
• Airflow 3 发布说明
• Ruff Airflow 规则

---

相关技能

• testing-dags：迁移后测试 DAG
• debugging-dags：排查迁移问题
• deploying-airflow：将迁移后的 DAG 部署到生产环境