如何在MWAA中将Airflow 2.5.1元数据迁移至3.0.x版本？

问题背景 #

需将MWAA环境中Airflow 2.5.1的元数据迁移至3.0.x版本，但Airflow 3.0.x已禁用直接数据库访问权限，此前通过settings创建会话并执行Insert SQL导入元数据的方式无法继续使用，而业务依赖历史运行记录。

尝试过的操作：编写DAG加载S3中存储的CSV格式元数据，通过settings创建会话并执行Insert SQL插入元数据库。该DAG在Airflow 2.10.3中可正常运行，但在3.0.6版本中抛出以下错误：

Task failed with exception: source="task"
RuntimeError: Direct database access via the ORM is not allowed in Airflow 3.0
File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 920 in run

File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1215 in _execute_task

File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 397 in wrapper

File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/providers/standard/operators/python.py", line 216 in execute

File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/providers/standard/operators/python.py", line 239 in execute_callable

File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/callback_runner.py", line 81 in run

File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/utils/session.py", line 100 in wrapper

File "/usr/local/lib/python3.12/contextlib.py", line 137 in enter

File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/utils/session.py", line 41 in create_session

File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/supervisor.py", line 262 in init

可行解决方案 #

1. 通过Airflow REST API导入元数据 #

Airflow 3.0.x提供了官方REST API用于元数据操作，可绕过直接数据库访问限制：

• 安装Airflow官方客户端库：pip install apache-airflow-client
• 在DAG的PythonOperator中编写代码，读取S3中的CSV数据，逐条调用对应API接口（例如POST /dags/{dag_id}/dagRuns创建DAG运行记录、POST /dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances创建任务实例记录）
• 确保MWAA的Airflow API已启用，且任务执行角色具备调用API的权限

2. 利用外部环境直接操作元数据库 #

Airflow运行时禁止ORM访问，但外部环境只要具备数据库权限即可操作：

• 配置MWAA元数据库（RDS）的安全组，允许EC2或AWS ECS实例访问
• 在EC2/ECS中运行自定义脚本，读取S3中的CSV数据，直接执行SQL插入元数据库
• 操作完成后恢复安全组配置，降低访问风险

3. 使用Airflow CLI结合临时环境迁移 #

• 在本地或EC2上搭建与MWAA版本一致的Airflow 3.0.x环境，配置连接至MWAA的元数据库
• 使用Airflow CLI的airflow db相关命令（如airflow db import）或自定义脚本通过CLI完成元数据导入
• 确保临时环境的依赖与MWAA完全匹配，避免版本兼容问题

内容的提问来源于stack exchange，提问作者Harsh Vardhan Chaudhary