如何快速将150列500万行的Pandas DataFrame导入SQL Server？

针对你500万行150列数据导入慢的问题，结合你的限制条件（无法使用BULK_INSERT），以下是经过验证的优化方案：

核心问题修正 #

你当前代码的最大浪费是分块读取后合并成大DataFrame再写入，完全失去了分块的意义——不仅占用大量内存，还拉长了写入时间。所有优化方案的前提都是：读一块，写一块，不缓存全量数据。

另外你遇到的COUNT field incorrect错误，本质是DataFrame列与SQL表列不匹配（名称、数量、顺序或数据类型），需先确保两者完全对齐。

方案一：优化Pandas to_sql写入 #

关键调整 #

1. 正确配置fast_executemany=True（是create_engine的参数，而非ODBC连接字符串内容）
2. 分块读写，避免全量加载
3. 严格对齐DataFrame与SQL表的列

import pandas as pd
from sqlalchemy import create_engine

# 正确构造连接引擎
conn_params = (
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=你的服务器地址;"
    "DATABASE=你的数据库;"
    "UID=用户名;"
    "PWD=密码;"
    "TrustedServerCertificate=Yes;"
    "autocommit=Yes"
)
engine = create_engine(
    f"mssql+pyodbc:///?odbc_connect={conn_params}",
    fast_executemany=True,
    isolation_level="AUTOCOMMIT"
)

# 分块读取并直接写入
chunk_size = 10000  # 150列建议用1万-2万的批次大小，避免ODBC参数超限
for chunk in pd.read_csv('file_path', chunksize=chunk_size, sep='|'):
    # 强制对齐列：按SQL表的列名和顺序筛选DataFrame列
    # chunk = chunk[['列1', '列2', ..., '列150']]
    chunk.to_sql(
        name='Table_Name',
        con=engine,
        if_exists='append',
        index=False,
        chunksize=chunk_size
    )

方案二：SQLAlchemy bulk_insert_mappings #

bulk_insert_mappings是SQLAlchemy专门针对批量插入优化的方法，比手动构造Insert()语句效率更高。

import pandas as pd
from sqlalchemy import create_engine, MetaData, Table
from sqlalchemy.orm import sessionmaker

# 连接引擎配置同方案一
conn_params = "DRIVER={ODBC Driver 18 for SQL Server};SERVER=你的服务器;DATABASE=你的库;UID=xxx;PWD=xxx;TrustedServerCertificate=Yes;autocommit=Yes"
engine = create_engine(
    f"mssql+pyodbc:///?odbc_connect={conn_params}",
    fast_executemany=True,
    isolation_level="AUTOCOMMIT"
)

# 加载目标表元数据
metadata = MetaData()
target_table = Table('Table_Name', metadata, autoload_with=engine)
Session = sessionmaker(bind=engine)

chunk_size = 10000
for chunk in pd.read_csv('file_path', chunksize=chunk_size, sep='|'):
    # 转换为与表列匹配的字典列表
    data_mappings = chunk.to_dict(orient='records')
    with Session() as session:
        session.bulk_insert_mappings(target_table, data_mappings)
        session.commit()

方案三：直接使用PyODBC fast_executemany #

绕过Pandas/SQLAlchemy的中间层，直接调用PyODBC的批量执行功能，减少额外开销。

import pandas as pd
import pyodbc

# 建立连接
conn_str = (
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=你的服务器;"
    "DATABASE=你的库;"
    "UID=用户名;"
    "PWD=密码;"
    "TrustedServerCertificate=Yes"
)
conn = pyodbc.connect(conn_str)
cursor = conn.cursor()
cursor.fast_executemany = True  # 开启快速批量执行

# 构造插入语句：确保列名与SQL表完全一致
sample_df = pd.read_csv('file_path', nrows=0, sep='|')
columns = ', '.join([f'[{col}]' for col in sample_df.columns])
placeholders = ', '.join(['?' for _ in sample_df.columns])
insert_sql = f"INSERT INTO Table_Name ({columns}) VALUES ({placeholders})"

# 分块写入
chunk_size = 10000
for chunk in pd.read_csv('file_path', chunksize=chunk_size, sep='|'):
    # 转换为元组列表匹配占位符
    data_tuples = [tuple(row) for row in chunk.values]
    cursor.executemany(insert_sql, data_tuples)
    conn.commit()

# 关闭连接
cursor.close()
conn.close()

额外优化建议 #

1. 数据类型预处理：提前将DataFrame的列转换为与SQL表匹配的数据类型（比如将字符串列设为合适长度，避免隐式转换开销）
2. 禁用约束：如果目标表有默认值、外键等约束，导入前临时禁用，导入后再启用（需确保数据合法性）
3. 批次大小调整：根据服务器性能和网络情况，微调chunk_size（150列建议1万-2万，避免单批次参数过多触发ODBC限制）

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