df.to_sql()插入过慢且fast_executemany=True报错的提速方案咨询

1. 修复fast_executemany的字符串截断错误 #

错误根源是fast_executemany模式下，驱动会根据DataFrame前几行推断参数长度，若后续行有更长字符串则触发截断。解决方式：

显式指定字符串列的SQL类型长度 #

确保dtypedict中所有字符串列都用sqlalchemy.VARCHAR指定足够大的长度，长度需覆盖DataFrame中该列的最大字符串长度。示例：

from sqlalchemy import VARCHAR

# 生成适配的dtypedict
dtypedict = {}
for col in df.columns:
    if df[col].dtype == object:
        # 计算列中最长字符串长度，留20%余量避免边界情况
        max_len = df[col].str.len().max()
        dtypedict[col] = VARCHAR(length=int(max_len * 1.2) if max_len else 50)
# 非字符串列按原有逻辑补充到dtypedict中

2. 优化插入速度 #

启用fast_executemany并优化引擎配置 #

修改connection.py的get_engine方法，直接启用fast_executemany并配置合理的连接池参数：

def get_engine(self) -> sqlalchemy.Engine:
    try:
        params = urllib.parse.quote_plus(self.connection_string)
        engine: sqlalchemy.Engine = sqlalchemy.create_engine(
            "mssql+pyodbc:///?odbc_connect=%s" % params,
            fast_executemany=True,
            pool_size=10,
            max_overflow=20
        )
        return engine
    except Exception as err:
        raise err

取消手动分块，用to_sql原生chunksize #

当前手动循环分块会增加多次to_sql调用的开销，直接使用to_sql的chunksize参数更高效：
修改write_data_to_stage中的插入逻辑：

# 删除手动分块的循环代码，替换为：
self.log.info('Start insertion to staging database')
start_time = time.time()
df.to_sql(
    str(row['DataTable']),
    engine,
    if_exists='append',
    index=False,
    dtype=dtypedict,
    chunksize=10000  # 根据内存情况调整，建议5000-20000
)
end_time = time.time()
elapsed_time = end_time - start_time
self.log.info(f'Dataframe insertion took {elapsed_time:.2f} seconds')

额外优化建议 #

• 确保使用ODBC Driver 17 for SQL Server或更高版本，旧驱动会限制fast_executemany的性能。
• 若数据库支持，可临时关闭目标表的索引/约束，插入完成后再重建，进一步提速。

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