Airflow执行db init报duplicate column name: operator错误咨询

我在学习Paul Crickard所著的《Data Engineering with Python》过程中，近期尝试安装配置Apache Airflow。完成安装后执行如下命令初始化数据库：

airflow db init

执行过程中抛出如下错误：

[2022-06-28 10:40:08,549] {db.py:1448} INFO - Creating tables
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade 13eb55f81627 -> 338e90f54d61, Add ``operator`` and ``queued_dttm`` to ``task_instance`` table
Traceback (most recent call last):
   File "/home/kien/.local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
self.dialect.do_execute(
   File "/home/kien/.local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
cursor.execute(statement, parameters)
sqlite3.OperationalError: duplicate column name: operator

我暂未找到有效解决方案，也不清楚编号为13eb55f81627的迁移文件存放位置。

• 操作系统：Ubuntu 20.04.04 LTS
• Python版本：3.8.6
• 核心依赖版本：apache-airflow==2.3.2、alembic==1.8.0、SQLAlchemy==1.4.9
  完整pip依赖列表通过pip freeze导出，包含celery、Flask、pandas等组件共200+个包。

为排查重复列名问题，我在sqlite命令行中执行了如下操作：

sqlite> .output text.txt
sqlite> .open airflow.db
sqlite> .dump

操作后发现原airflow.db被清空，导出的text.txt重命名为airflow.sql后仅包含如下内容：

PRAGMA foreign_keys=OFF;
BEGIN TRANSACTION;
COMMIT;

此时我重新执行初始化命令：

airflow db init

执行过程未抛出任何错误，数据库初始化成功。

• 上述操作解决报错的具体原理是什么？
• 第一次初始化失败的根本原因是什么？

第一次初始化失败的根本原因 #

报错本质是Alembic迁移状态与实际表结构不一致导致的。
Airflow使用Alembic管理数据库schema变更，执行airflow db init时会按版本号顺序依次执行所有迁移脚本，每执行完一个迁移就会把对应版本号写入数据库内的alembic_version表，标记迁移完成。
第一次执行初始化时，大概率出现过中途中断、进程崩溃、权限不足等异常：跑到338e90f54d61号迁移（给task_instance表加operator和queued_dttm列）时，operator列已经被实际写入表结构，但Alembic没来得及更新版本记录就中断了。后续重新执行初始化时，Alembic查版本表发现这个迁移还没标记完成，就再次执行加列操作，SQLite检测到同名列已存在，直接抛出重复列名错误。

操作生效的原理 #

在sqlite命令行的操作顺序存在逻辑问题，相当于手动清除了状态异常的残留坏库：
进入sqlite cli时默认连接的是内存临时空库，先执行.output text.txt设置了输出文件，之后才执行.open airflow.db加载本地数据库文件，这个操作顺序下执行.dump时导出的是空库结构，同时操作过程中异常覆盖/清空了本地原本状态不一致的airflow.db文件——效果等同于手动删除损坏的旧数据库文件。
第二次执行airflow db init时，程序检测到数据库是空的，没有任何残留的半完成表结构和错误的版本记录，就从头完整执行了所有迁移步骤，自然不会再报重复列的错误。

补充说明 #

查找的13eb55f81627号迁移文件，存放在本地Airflow安装路径的迁移目录下，对应本机的完整路径为：
/home/kien/.local/lib/python3.8/site-packages/airflow/migrations/versions/，目录内文件名开头就是对应迁移的版本号，可以直接查看。

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