如何高效批量分块将CSV数据插入SQL Server表？解决executemany报错

问题背景 #

原本采用逐行插入的方式将CSV数据导入SQL Server，速度极慢；尝试用executemany批量插入时，遭遇SQL Server Statement(s) could not be prepared(8180)语法错误。

原始逐行插入代码 #

import pandas as pd
df = pd.read_csv(r'd:/data_file.csv')
for index, row in df.iterrows():
   cursor.execute("insert into <table name>"
   "(Name, phone, address)"
   "VALUES(?, ?, ?)",
    (row['FirstandLastName'], row['xxx'], row['11 adison ln']))

出错的批量尝试代码 #

query=f"""insert into <table name>"
   "(Name, phone, address)"
   "VALUES(?, ?, ?)"""
values_list=[tuple(row) for _, row in df.iterrows()]

try:
   cursor.executemany(query, values_list)
except Exception as e:
   print(e)

报错信息 #

SQL Server Statement(s) could not be prepared(8180)

错误修复：修正executemany的语法问题 #

你写的query字符串存在语法错误：f-string中多了一个多余的双引号，导致SQL语句格式混乱。同时iterrows效率偏低，建议改用itertuples生成数据列表。修复后的代码：

# 修正SQL语句格式
query = """INSERT INTO <table name> (Name, phone, address) VALUES (?, ?, ?)"""

# 用itertuples生成元组列表，比iterrows更高效
values_list = [(row.FirstandLastName, row.xxx, row['11 adison ln']) for row in df.itertuples(index=False)]

try:
    cursor.executemany(query, values_list)
    conn.commit()  # 别忘了提交事务
except Exception as e:
    print(e)

更高效的批量导入方案 #

如果数据量较大，以下方法能进一步提升导入速度：

1. 使用pandas内置to_sql方法 #

pandas的to_sql支持分块批量插入，配合SQLAlchemy引擎使用效率更高：

from sqlalchemy import create_engine

# 创建SQL Server连接引擎（替换为你的连接信息）
engine = create_engine('mssql+pyodbc://<用户名>:<密码>@<服务器>/<数据库>?driver=ODBC+Driver+17+for+SQL+Server')

# 分块导入，chunksize根据内存调整（建议1000-10000）
df.to_sql(
    '<table name>',
    engine,
    if_exists='append',  # 可选：replace（替换表）/fail（表存在则报错）
    index=False,
    chunksize=1000
)

2. 使用SQL Server原生BULK INSERT #

这是速度最快的方案之一，前提是SQL Server服务账户能直接访问CSV文件路径：

bulk_query = """
BULK INSERT <table name>
FROM 'D:\data_file.csv'
WITH (
    FIELDTERMINATOR = ',',    # CSV字段分隔符
    ROWTERMINATOR = '\n',     # 行分隔符
    FIRSTROW = 2,             # 若CSV有表头，从第2行开始导入
    TABLOCK                   # 锁定表提升导入性能
)
"""
cursor.execute(bulk_query)
conn.commit()

3. 开启pyodbc的fast_executemany #

pyodbc 4.0.19+版本支持fast_executemany，开启后会大幅优化executemany的执行效率：

# 创建连接时开启fast_executemany
conn = pyodbc.connect('<你的连接字符串>', fast_executemany=True)
cursor = conn.cursor()

query = """INSERT INTO <table name> (Name, phone, address) VALUES (?, ?, ?)"""
values_list = [(row.FirstandLastName, row.xxx, row['11 adison ln']) for row in df.itertuples(index=False)]

cursor.executemany(query, values_list)
conn.commit()

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