SQLAlchemy+Pandas to_sql插入MS SQL时因系统视图查询挂起的解决问询

我之前也碰到过一模一样的情况——Pandas的to_sql()在执行插入前，会自动查询INFORMATION_SCHEMA.TABLES来确认目标表是否存在，而当数据库里有未提交的建表事务时，这个查询会被阻塞，直接导致进程挂起。结合你用if_exists='append'的场景，这里有两个可行的解决方案：

方法一：自定义SQLTable跳过表存在性检查 #

既然你用的是append模式，说明你已经确认目标表肯定存在，那完全可以跳过这个多余的存在性检查，避免触发那个会被阻塞的查询。

具体步骤和代码如下：

import pandas as pd
from pandas.io.sql import SQLTable, SQLAlchemyConnection

# 自定义SQLTable类，重写exists方法直接返回True
class SkipCheckSQLTable(SQLTable):
    def exists(self):
        return True  # 跳过表存在性检查

# 自定义连接类，使用我们的SkipCheckSQLTable
class SkipCheckSQLAlchemyConnection(SQLAlchemyConnection):
    def _get_sql_table(self, table_name, df, dtype, schema):
        return SkipCheckSQLTable(
            table_name,
            self,
            frame=df,
            dtype=dtype,
            schema=schema,
            if_exists='append',
            index=False
        )

# 创建引擎（建议加上fast_executemany提升插入速度）
engine = sqlalchemy.create_engine(
    "mssql+pyodbc:///?odbc_connect=%s" % params,
    fast_executemany=True
)

with engine.connect() as connection:
    # 实例化自定义连接
    custom_conn = SkipCheckSQLAlchemyConnection(engine)
    # 执行插入，最好指定dtype避免其他潜在的反射操作
    df.to_sql(
        name=my_table,
        con=custom_conn,
        if_exists='append',
        index=False,
        dtype={
            'column1': sqlalchemy.types.String(50),
            'column2': sqlalchemy.types.Integer(),
            # 替换成你的表实际字段和类型
        }
    )

方法二：直接用SQLAlchemy Core执行批量插入 #

完全绕开Pandas的to_sql()逻辑，直接用SQLAlchemy的Core层构造插入语句，这样根本不会触发任何表结构查询，从根源上解决阻塞问题。

代码示例：

import sqlalchemy as sa
from sqlalchemy import Table, Column, String, Integer, MetaData

# 手动定义目标表的结构（必须和数据库中的表完全匹配）
metadata = MetaData()
target_table = Table(
    my_table,
    metadata,
    Column('column1', String(50)),
    Column('column2', Integer()),
    # 补充你的其他字段...
    schema=None  # 如果表在特定schema下，这里指定
)

# 创建引擎
engine = sa.create_engine(
    "mssql+pyodbc:///?odbc_connect=%s" % params,
    fast_executemany=True
)

with engine.connect() as connection:
    # 把DataFrame转成字典列表
    data_records = df.to_dict('records')
    # 构造插入语句并执行
    insert_stmt = target_table.insert().values(data_records)
    connection.execute(insert_stmt)
    # 根据你的引擎配置，可能需要手动提交
    connection.commit()

一些注意点 #

• 方法一只能在你100%确认目标表已经存在的情况下使用，否则会抛出表不存在的错误。
• 方法二中，手动定义表结构时要保证和数据库中的表完全一致，包括字段名、类型、长度等，否则插入会失败。如果不想手动写，也可以在数据库没有阻塞的时候提前反射表结构保存下来，后续复用。
• 开启fast_executemany=True对MS SQL的批量插入速度提升非常明显，强烈建议加上。

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