mcpField
M
migrating-airflow-2-to-3

migrating-airflow-2-to-3

github.com/necatiarslan数据与数据库4448 次安装21 周前更新
代码仓库

关于 migrating-airflow-2-to-3

用于将 Airflow 2.x DAG 代码迁移到 Airflow 3.x 的技能，涵盖代码调整、迁移检查和分步验证。

主要功能

  1. 01运行 Ruff Airflow 迁移规则
  2. 02手动迁移检查清单
  3. 03更新导入和上下文
  4. 04元数据数据库访问指导
  5. 05分步迁移工作流

适用场景

  1. 01查找已弃用的 Airflow 代码
  2. 02替换直接 ORM 访问
  3. 03更新 DAG 导入和上下文
  4. 04规划并验证迁移改动

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/necatiarslan/airflow-vscode-extension --skill migrating-airflow-2-to-3

安装后在 Claude Code 中输入/migrating-airflow-2-to-3确认已加载。

详细介绍

Airflow 2 到 3 迁移

此技能可帮助将 Airflow 2.x DAG 代码迁移到 Airflow 3.x，重点关注代码变更（导入、运算符、Hook、上下文、API 用法）。

重要提示：迁移到 Airflow 3 之前，强烈建议先升级到 Airflow 2.11，然后再升级到至少 Airflow 3.0.11（最好是 3.1）。

迁移概览

  1. 运行 Ruff Airflow 迁移规则：
    • ruff check --preview --select AIR --fix --unsafe-fixes .
  2. 使用手动检查清单扫描剩余问题：
    • 直接访问元数据数据库
    • 旧版导入
    • 调度和上下文变更
    • XCom Pickling
    • 从数据集迁移到资产
    • REST API 和身份验证
    • 插件和文件路径
  3. 按文件和问题类型规划变更
  4. 逐步实施变更并重新运行 Ruff
  5. 说明变更并建议进行测试

架构和元数据数据库访问

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

  • Worker 不再直接连接元数据数据库
  • 任务代码通过 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.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

上下文键变更

已移除的键替代项
execution_datecontext["dag_run"].logical_date
tomorrow_ds / yesterday_ds使用 macros.ds_add
triggering_dataset_eventstriggering_asset_events
templates_dictcontext["params"]

默认行为变更

设置Airflow 2 默认值Airflow 3 默认值
scheduletimedelta(days=1)None
catchupTrueFalse

资源

评论

支持 Markdown