Apache Airflow 版本升级完整指南：从数据库迁移备份到故障恢复

Apache Airflow 版本升级完整指南从数据库迁移备份到故障恢复【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowApache Airflow 的每一次版本升级都伴随元数据库Metadata Databaseschema 的演进升级过程的核心风险与关键操作几乎都集中在数据库迁移上。本文基于官方 Upgrading Airflow to a newer version 文档结合仓库中 db_command.py 等 CLI 实现源码系统讲解 Airflow 升级的完整流程为什么必须升级、升级前如何备份、何时需要手动迁移、如何离线生成 SQL 迁移脚本、以及升级失败时如何修复 MySQL 编码问题与清理被迁移的异常数据帮助你掌握一套可复制、可回滚的升级实战方案。为什么必须升级数据库迁移是刚性要求Airflow 的元数据库存储 DAG、任务实例Task Instance、XCom、调度状态等全部核心状态。新版本可能包含数据库迁移Database Migration脚本因此升级到新版本后必须运行airflow db migrate将数据库 schema 迁移到新版本要求的结构。不要担心重复执行——即使当前版本没有任何待执行的迁移再次运行该命令也是完全安全的。从源码看迁移命令最终通过 Alembic 完成db_manager.py 中BaseDBManager.upgradedb()会先读取当前 revision再调用 Alembic 的command.upgrade(config, revisionto_revision or heads, sqlshow_sql_only)Alembic 自身会追踪已执行的迁移版本幂等地跳过已应用过的变更。两个版本之间到底改了什么每个 Airflow 版本包含的具体变更统一收录在 release notes 中。升级前建议对照发布说明逐一确认是否有新增的核心功能、行为变更significant changes、已知问题以及是否引入了新的数据库迁移脚本。完整的迁移脚本清单可查阅 migrations-ref。升级准备先备份元数据库官方强烈建议在任何迁移之前备份元数据库这是整个升级过程中最重要的一道保险如果没有数据库的热备份hot backup能力应先关闭所有 Airflow 实例再执行备份以保证备份数据的一致性如果不备份就直接迁移一旦迁移中途失败例如 CLI 与数据库之间的网络连接在迁移过程中断开数据库会处于半迁移half-migrated状态此时从备份恢复并重试迁移往往是唯一简单可行的出路。备份的对象是元数据库本身具体工具取决于你使用的数据库类型PostgreSQL 的pg_dump、MySQL 的mysqldump等Airflow 只负责消费sql_alchemy_conn指向的数据库连接。什么时候需要升级手动与自动的判定是否需要在升级过程中手动执行数据库迁移取决于你的部署方式virtualenv 或 Docker 容器自定义部署通常需要在升级过程中手动运行airflow db migrateHelm Chart 部署如果启用了 post-upgrade 钩子新软件安装完成后数据库升级会自动进行。仓库中的 migrate-database-job.yaml 就是官方 Helm Chart 的数据库迁移 Job它由values.yaml中的migrateDatabaseJob.enabled控制默认开启该 Job 会在 Airflow 组件webserver、scheduler、worker启动前完成airflow db migrate从而保证各组件以正确的 schema 启动Airflow-As-A-Service托管服务在界面上选择升级 Airflow 时服务商会自动完成数据库升级。如何升级安装新版本并迁移数据库第一步重装 Airflow 到目标版本升级的第一步是重新安装 Apache Airflow并指定想要的新版本。对于 PyPI 包在你的环境中用目标版本重新执行pip install并务必携带 constraints 约束文件以固定依赖版本、避免依赖解析漂移pip install apache-airflow[celery]目标版本 \ --constraint https://raw.githubusercontent.com/apache/airflow/constraints-目标版本/constraints-3.10.txt注意上例中3.10是你的 Python 版本需按实际环境替换URL 为约束文件的外部获取途径仓库内的安装细节与更多升级场景见 installing-from-pypi。详细的安装与升级场景包括 constraints 机制、依赖管理建议参见 installing-from-pypi。第二步手动执行数据库迁移在安装了新版本的环境中运行airflow db migrate该命令可以在虚拟环境中运行也可以在拥有 Airflow CLI 与数据库访问权的容器中运行。CLI 的完整用法见 usage-cli。从 CLI 实现看db migrate命令cli_config.py 中的DB_COMMANDS定义支持以下参数参数说明-n, --to-version可选升级到的 Airflow 版本号-r, --to-revision可选仅执行到指定 Alembic revision-s, --show-sql-only只打印 SQL 脚本不实际执行迁移离线迁移--from-version可选生成 SQL 时指定的起始版本--from-revision可选生成 SQL 时指定的起始 Alembic revision-m, --use-migration-files使用迁移文件而非 ORM 生成 schema用于校验迁移文件与 ORM 模型一致性-v, --verbose详细输出这些参数定义在 cli_config.py由 db_command.py 的run_db_migrate_command()消费。值得注意的校验逻辑--to-revision与--to-version不能同时提供--from-version与--from-revision也不能同时提供--from-version/--from-revision只能与--show-sql-only组合使用——因为实际执行迁移时必须从数据库当前的 Alembic revision 出发只有离线生成 SQL 时才需要显式指定起点migratedb()还会强制校验--from-version必须大于等于2.0.0见 db_command.py。离线 SQL 迁移脚本先预览再执行如果你希望离线运行升级脚本例如在割接窗口之外先评审 SQL可以使用-s/--show-sql-only参数它会打印将要执行的 SQL 语句而不实际提交。配合--from-version起始版本与-n/--to-version目标版本可以精确生成指定版本区间的迁移 SQL。该特性自 Airflow 2.0.0 起支持 Postgres 与 MySQL。Airflow 2.7.0 及以上版本的示例用法airflow db migrate -s --from-version 2.4.3 -n 2.7.3 airflow db migrate --show-sql-only --from-version 2.4.3 --to-version 2.7.3执行时会输出日志Generating sql for upgrade -- upgrade commands will *not* be submitted.明确告知仅生成 SQL、不会执行见 db_command.py。重要弃用提示airflow db upgrade自 Airflow 2.7.0 起已被airflow db migrate取代并弃用请使用新命令。处理迁移问题MySQL 数据库编码错误修复如果你使用 Airflow 1.10 时代创建的数据库无论是手工创建还是由旧版 MySQL 创建迁移到新版时可能因为原始字符集问题而失败报出诸如 key size too big、missing indexes 之类的奇怪错误。原因如下MySQL 8 推荐使用utf8mb4字符集与utf8mb4_bin排序规则但 MySQL 对索引键大小有限制。在utf8mb4下Airflow 的索引键可能过大而超出 MySQL 的处理能力。因此Airflow 强制所有 ID 键使用utf8字符集在 MySQL 8 中等价于utf8mb3从而压缩索引尺寸。以下步骤应在尝试迁移之前完成如果你清楚自己在做什么也可以自行处理。正式操作前建议先熟悉 Airflow 内部数据库结构见 database-erd-ref与迁移清单见 migrations-ref。1. 备份数据库先做备份确保出错时可以恢复。2. 检查哪些表需要修复SHOW CREATE TABLE task_reschedule; SHOW CREATE TABLE xcom; SHOW CREATE TABLE task_fail; SHOW CREATE TABLE rendered_task_instance_fields; SHOW CREATE TABLE task_instance;务必保存输出最后一步重建外键时需要用到。你的dag_id、run_id、task_id、key列应显式设置为utf8或utf8mb3字符集类似task_id varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL, # 正确或task_id varchar(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin NOT NULL, # 正确存在问题的字段形态包括完全没有编码声明task_id varchar(250), # 错误 !!仅 collation 为 utf8mb4task_id varchar(250) COLLATE utf8mb4_unicode_ci DEFAULT NULL, # 错误 !!字符集与排序规则均为 utf8mb4task_id varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL, # 错误 !!你需要修复所有字符集/排序规则错误的字段。3. 删除需要修改表的外键索引只需删除将要修改的表的不需要全部删除最后一步会重建它们ALTER TABLE task_reschedule DROP FOREIGN KEY task_reschedule_ti_fkey; ALTER TABLE xcom DROP FOREIGN KEY xcom_task_instance_fkey; ALTER TABLE task_fail DROP FOREIGN KEY task_fail_ti_fkey; ALTER TABLE rendered_task_instance_fields DROP FOREIGN KEY rtif_ti_fkey;4. 将 ID 字段修改为正确的字符集/编码仅对编码错误的字段执行以下是所有可能需要用到的命令ALTER TABLE task_instance MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_reschedule MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE rendered_task_instance_fields MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE rendered_task_instance_fields MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_fail MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_fail MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE sla_miss MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE sla_miss MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_map MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_map MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_map MODIFY run_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE xcom MODIFY task_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE xcom MODIFY dag_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE xcom MODIFY run_id VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE xcom MODIFY key VARCHAR(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;5. 重建第 3 步删除的外键对删除的每个索引重复此操作。注意不同 Airflow 版本的索引可能略有差异例如map_index是在 2.3.0 加入的但只要保留第 2 步的SHOW CREATE TABLE输出就能从中找到正确的CONSTRAINT_NAME与CONSTRAINT# 此处请从 SHOW CREATE TABLE 输出中复制语句 ALTER TABLE TABLE ADD CONSTRAINT CONSTRAINT_NAME CONSTRAINT完成上述修复后数据库即可正常迁移到新版本。升级后的警告被迁移的异常数据通常你只需要成功执行airflow db migrate即可。但在某些情况下迁移会发现数据库中陈旧、疑似损坏的数据并将其移到单独的表。此时 webserver UI 会显示类似警告Airflow found incompatible data in the table in the metadatabase, and has moved them to during the database migration to upgrade. Please inspect the moved data to decide whether you need to keep them, and manually drop the table to dismiss this warning.Airflow 在元数据库的 原表 中发现了不兼容数据并在升级迁移期间将其移到了 新表。请检查被移动的数据以决定是否保留然后手动删除 新表 以消除该警告。出现该消息意味着部分数据已损坏很可能是某些 bug 遗留的数据在 Airflow 中本就不可见、无用处多数情况下可安全删除。除非你有审计或历史留存等特殊需求否则删除这些数据通常是最佳选择。检查与删除被迁移数据的多种方式使用自己的数据库工具通常是图形化工具可以直接 drop、rename 该表或将其移动到其他数据库没有现成工具时使用airflow db shell命令进入数据库 shell。从源码看该命令会基于sql_alchemy_conn自动选择客户端db_command.pyMySQL 使用mysql客户端并自动将连接 URL 中的 SSL 等参数写入临时my.cnf选项文件SQLite 使用sqlite3PostgreSQL 使用psql通过环境变量注入连接信息。检查表内容SELECT * FROM table;删除表DROP TABLE table;Kubernetes 环境下删除表的步骤进入任一 Airflow Podwebserver 或 schedulerkubectl exec -it your-webserver-pod python在 Python shell 中执行from airflow.settings import Session session Session() session.execute(DROP TABLE _airflow_moved__2_2__task_instance) session.commit()请将示例中的表名替换为警告信息中打印的实际表名。迁移最佳实践先演练再上线数据库迁移耗时取决于库的大小与实际迁移内容如果历史数据长、数据库大建议先复制一份数据库做一次测试迁移评估迁移耗时通常Major大版本升级耗时更长——新特性往往需要重构数据库结构例如新增核心表、调整约束等在演练过程中结合--show-sql-only离线生成 SQL 评审可以提前发现大表重建、锁表等风险点。小结Airflow 升级的完整闭环可以概括为四步读发布说明 → 备份元数据库 → 升级安装包并执行airflow db migrate→ 处理迁移警告。遇到 MySQL 编码问题时按备份 → 检查 → 删外键 → 改编码 → 重建外键的顺序修复遇到被迁移的异常数据时用airflow db shell或 Kubernetes exec 检查并清理。掌握了这些操作与源码层面的机制无论是小版本迭代还是跨大版本升级都能做到有备无患、可回滚、可验证。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考