Python实现SQL Server同库表数据复制速度过慢求优化方案

问题概要 #

需将同一SQL Server数据库中某表的数百万条数据复制到另一张表，当前实现每100条记录耗时4.5秒，速度无法满足需求。已尝试以下方案但效果不佳：

• 逐条循环插入：速度极慢
• SQLAlchemy批量插入：仅小幅提升，调整chunk_size无明显改善
• 并行处理：触发驱动错误：Error inserting data: ('IM002', '[IM002] [Microsoft][ODBC Driver Manager] Data source name not found and no default driver specified (0) (SQLDriverConnect)')

最优解决思路 #

方案1：数据库原生SQL操作（最快） #

直接在SQL Server内执行数据复制，无需通过Python传输数据，这是同库复制的最优选择：

-- 场景1：目标表不存在，自动创建并复制数据
SELECT {sourcecolumns}
INTO PRODPARMUPDATE_tst
FROM {tablename}
-- 可添加过滤条件，比如 WHERE LocID=3;

-- 场景2：目标表已存在，插入数据
INSERT INTO PRODPARMUPDATE_tst ({sourcecolumns})
SELECT {sourcecolumns}
FROM {tablename}
-- 可添加过滤条件，比如 WHERE LocID=3;

核心优势：完全由数据库引擎处理，避免Python与数据库之间的网络/IO开销，速度比Python批量插入快数倍甚至数十倍。

方案2：优化Python批量插入（需中间处理时使用） #

如果必须通过Python处理数据（如字段转换、逻辑校验），可优化现有SQLAlchemy实现：

import pandas as pd
from sqlalchemy import create_engine
import datetime

# 建立引擎，确保fast_executemany=True开启
connection_string = f'mssql+pyodbc://{connection_info[2]}:{connection_info[3]}@{connection_info[0]}/{connection_info[1]}?driver=ODBC+Driver+17+for+SQL+Server'
sqlalchemy_engine = create_engine(connection_string, fast_executemany=True)

# 获取映射配置
mapping_detail_query = """
SELECT TableName, MappingName, SourceColumns, "Filter"
FROM MappingTable
WHERE LocID=3
"""
mapping_detail = execute_sqlite_query(sqlite_conn, mapping_detail_query)[0]
tablename, mappingname, sourcecolumns, _filter = mapping_detail

# 大批次读取并写入
chunk_size = 10000  # 根据内存调整，建议1万-5万条/批次
for chunk_df in pd.read_sql_query(f"SELECT {sourcecolumns} FROM {tablename}", sqlalchemy_engine, chunksize=chunk_size):
    # 使用to_sql批量写入，method='multi'配合fast_executemany提升效率
    chunk_df.to_sql(
        name='PRODPARMUPDATE_tst',
        con=sqlalchemy_engine,
        if_exists='append',
        index=False,
        method='multi'
    )
    current_datetime = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"批次插入完成 - {current_datetime}")

优化要点：

• 大幅提升chunk_size，减少数据库交互次数
• 依赖SQLAlchemy的to_sql方法，避免手动拼接SQL的风险，同时利用fast_executemany=True优化批量插入性能
• 无需手动维护累加队列，代码更简洁可靠

方案3：修复并行处理的驱动错误 #

若需并行处理，需解决线程内连接参数传递问题，修复后的实现：

import pandas as pd
from sqlalchemy import create_engine
from concurrent.futures import ThreadPoolExecutor
import pyodbc
import datetime

def insert_chunk(data_chunk, conn_str, target_table, columns):
    try:
        # 每个线程独立创建数据库连接
        conn = pyodbc.connect(conn_str)
        cursor = conn.cursor()
        placeholders = ', '.join(['?' for _ in columns.split(',')])
        insert_query = f"""
        INSERT INTO {target_table} ({columns})
        VALUES ({placeholders})
        """
        cursor.executemany(insert_query, data_chunk)
        conn.commit()
        current_datetime = datetime.datetime.now().strftime("%H:%M:%S")
        print(f"批次插入完成 - {current_datetime}")
    except Exception as e:
        print(f"插入失败: {e}")
    finally:
        # 确保连接关闭
        if 'conn' in locals():
            conn.close()

# 主逻辑
connection_string = f'mssql+pyodbc://{connection_info[2]}:{connection_info[3]}@{connection_info[0]}/{connection_info[1]}?driver=ODBC+Driver+17+for+SQL+Server'
sqlalchemy_engine = create_engine(connection_string, fast_executemany=True)

# 获取映射配置
mapping_detail_query = """
SELECT TableName, MappingName, SourceColumns, "Filter"
FROM MappingTable
WHERE LocID=3
"""
mapping_detail = execute_sqlite_query(sqlite_conn, mapping_detail_query)[0]
tablename, mappingname, sourcecolumns, _filter = mapping_detail

# 拆分数据为大批次
chunk_size = 10000
data_chunks = []
for chunk_df in pd.read_sql_query(f"SELECT {sourcecolumns} FROM {tablename}", sqlalchemy_engine, chunksize=chunk_size):
    data_chunks.append(chunk_df.values.tolist())

# 并行插入
with ThreadPoolExecutor(max_workers=4) as executor:
    executor.map(
        lambda chunk: insert_chunk(chunk, connection_string, 'PRODPARMUPDATE_tst', sourcecolumns),
        data_chunks
    )

修复说明：

• 将连接字符串、目标表、列名作为参数传入线程函数，避免线程变量作用域问题
• 每个线程独立创建和关闭连接，避免连接共享冲突
• 添加finally块确保连接资源被释放

总结 #

优先选择方案1的数据库原生操作，这是同库数据复制的最快方式；若需Python介入数据处理，使用方案2的优化批量插入；并行方案仅适合单线程性能不足且必须保留中间处理逻辑的场景，注意修复连接参数传递问题。

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