气流操作错误 sqlalquemy (sqlite3.OperationalError)答案

【问题标题】：Airflow Operational error sqlalquemy (sqlite3.OperationalError)气流操作错误 sqlalquemy (sqlite3.OperationalError)
【发布时间】：2022-06-16 19:17:47
【问题描述】：

我正在尝试安装气流，当我尝试创建数据库或独立运行时出现此错误。我尝试更改 sqlalchemy 的版本，但没有成功。

操作系统：manjaro | sqlalchemy 版本：1.4 | 气流版本：2.3.0

standalone | Starting Airflow Standalone
standalone | Checking database is initialized
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 
sqlite3.OperationalError: duplicate column name: operator

The above exception was the direct cause of the following exception:

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) duplicate column name: operator
[SQL: ALTER TABLE task_instance ADD COLUMN operator VARCHAR(1000)]
(Background on this error at: http://sqlalche.me/e/14/e3q8)

有什么办法解决吗？

【问题讨论】：

你可以在这个链接stackoverflow.com/questions/56988054/…找到答案

标签： python python-3.x sqlalchemy airflow

【解决方案1】：

我发现我检查了'Shi Chen'指出的代码文件，两个文件是造成这种行为的原因。

33ae817a1ff4_add_kubernetes_resource_checkpointing.py
86770d1215c0_add_kubernetes_scheduler_uniqueness.py

这两个文件都是使用 alembic 和 sqlalchemy 库的迁移文件，我发现以下 sqlalchemy 代码写在文件 33ae817a1ff4_add_kubernetes_resource_checkpointing.py 中

def upgrade():

    columns_and_constraints = [
        sa.Column("one_row_id", sa.Boolean, server_default=sa.true(), primary_key=True),
        sa.Column("resource_version", sa.String(255))
    ]

    conn = op.get_bind()

    # alembic creates an invalid SQL for mssql dialect
    if conn.dialect.name not in ('mssql'):
        columns_and_constraints.append(sa.CheckConstraint("one_row_id", name="kube_resource_version_one_row_id"))

    table = op.create_table(
        RESOURCE_TABLE,
        *columns_and_constraints
    )

    op.bulk_insert(table, [
        {"resource_version": ""}
    ])

被解释为以下不正确的 SQL 查询

CREATE TABLE  kube_resource_version (one_row_id BOOL NOT NULL DEFAULT
true,  resource_version VARCHAR(255), PRIMARY KEY (one_row_id), 
CONSTRAINT kube_resource_version_one_row_id CHECK (one_row_id),  CHECK(one_row_id IN (0, 1))

相反，SQL 查询应该是这样的

CREATE TABLE  kube_resource_version (one_row_id BOOL NOT NULL DEFAULT
true,  resource_version VARCHAR(255), PRIMARY KEY (one_row_id), 
CONSTRAINT kube_resource_version_one_row_id CHECK (one_row_id IN (0, 1)))

“skadya”提供的链接很有帮助，在对上述两个文件的代码进行更改后，我让系统正常工作。

您只需将以下代码更改为

if conn.dialect.name not in ('mssql'):
columns_and_constraints.append(
sa.CheckConstraint("one_row_id", name="kube_resource_version_one_row_id"))

到

if conn.dialect.name not in ('mssql', 'mysql'):
columns_and_constraints.append(
sa.CheckConstraint("one_row_id", name="kube_resource_version_one_row_id")

【讨论】：