Airflow升级至2.10.x时SQL Server数据库迁移失败求助

问题分析 #

你遇到的错误核心原因是：SQL Server的TIMESTAMP并非日期时间类型，而是用于行版本控制的特殊标识符，单表仅允许存在一个该类型列。但Airflow 2.10.x的数据库迁移脚本错误地将task_instance_history表中的多个日期时间字段（如start_date、end_date等）映射为SQL Server的TIMESTAMP类型，直接违反了SQL Server的约束规则。

可行解决方案 #

方案1：修改迁移脚本字段类型 #

1. 定位迁移脚本文件：
   找到Airflow安装目录下对应创建task_instance_history表的迁移文件，路径通常为 <你的Airflow安装路径>/migrations/versions/，文件名格式类似 [版本哈希]_create_task_instance_history.py（对应2.10.0版本的迁移逻辑）。
2. 替换字段类型：
   打开脚本，将所有定义为TIMESTAMP的字段，替换为SQL Server支持的高精度日期时间类型DATETIME2(6)（该类型匹配Airflow对时间精度的需求）。
3. 重新执行迁移：

   airflow db migrate
   

方案2：手动创建表并跳过迁移 #

1. 修正建表SQL：
   复制错误提示中的CREATE TABLE语句，将所有TIMESTAMP替换为DATETIME2(6)，示例修正后的SQL如下：

   CREATE TABLE task_instance_history (
           id INTEGER NOT NULL IDENTITY,
           task_id VARCHAR(250) NOT NULL,
           dag_id VARCHAR(250) NOT NULL,
           run_id VARCHAR(250) NOT NULL,
           map_index INTEGER NOT NULL DEFAULT -1,
           try_number INTEGER NOT NULL,
           start_date DATETIME2(6) NULL,
           end_date DATETIME2(6) NULL,
           duration FLOAT NULL,
           state VARCHAR(20) NULL,
           max_tries INTEGER NULL DEFAULT -1,
           hostname VARCHAR(1000) NULL,
           unixname VARCHAR(1000) NULL,
           job_id INTEGER NULL,
           pool VARCHAR(256) NOT NULL,
           pool_slots INTEGER NOT NULL,
           queue VARCHAR(256) NULL,
           priority_weight INTEGER NULL,
           operator VARCHAR(1000) NULL,
           custom_operator_name VARCHAR(1000) NULL,
           queued_dttm DATETIME2(6) NULL,
           queued_by_job_id INTEGER NULL,
           pid INTEGER NULL,
           executor VARCHAR(1000) NULL,
           executor_config VARBINARY(max) NULL,
           updated_at DATETIME2(6) NULL,
           rendered_map_index VARCHAR(250) NULL,
           external_executor_id VARCHAR(250) NULL,
           trigger_id INTEGER NULL,
           trigger_timeout DATETIME NULL,
           next_method VARCHAR(1000) NULL,
           next_kwargs NVARCHAR(max) NULL,
           task_display_name VARCHAR(2000) NULL,
           CONSTRAINT task_instance_history_pkey PRIMARY KEY (id),
           CONSTRAINT task_instance_history_ti_fkey FOREIGN KEY(dag_id, task_id, run_id, map_index) REFERENCES task_instance (dag_id, task_id, run_id, map_index) ON DELETE CASCADE ON UPDATE CASCADE,
           CONSTRAINT task_instance_history_dtrt_uq UNIQUE (dag_id, task_id, run_id, map_index, try_number)
   )
   

2. 在SQL Server中执行上述修正后的SQL，完成表创建。
3. 标记迁移版本为已完成：
   找到对应迁移脚本的版本哈希（即文件名开头的字符串），在SQL Server的alembic_version表中插入该版本号，示例SQL：

   INSERT INTO alembic_version (version_num) VALUES ('[迁移版本哈希]');
   

4. 继续执行后续迁移：

   airflow db migrate
   

方案3：等待官方修复 #

这是Airflow针对SQL Server后端的兼容性bug，后续官方版本大概率会修复该类型映射错误，你可以关注Airflow的更新日志，待修复版本发布后直接升级即可。

内容的提问来源于stack exchange，提问作者RedBlue