PYODBC批量插入SQL Server提速及连接问题咨询

1. 优化批量插入的块大小 #

当前1万条的块大小未必是最优值，建议测试不同的块大小（比如5万、10万条），找到性能平衡点：

• 块太小会增加事务提交次数和网络交互次数，浪费开销
• 块太大可能导致内存占用过高、执行超时或连接断开

2. 使用表值参数（TVP）替代executemany #

TVP是SQL Server对批量插入的原生支持，性能比fast_executemany=True更优，因为它是一次性将结构化数据发送到服务器，而非批量执行参数化语句。
步骤：

1. 在SQL Server创建用户定义表类型：

CREATE TYPE dbo.YourTableType AS TABLE (
    Column1 INT,
    Column2 VARCHAR(50),
    -- 匹配你的目标表结构
);

2. pyodbc代码实现：

import pyodbc
import time

# 假设chunk是列表的列表，每个子列表对应一行数据
def insert_with_tvp(connection_string, chunk, table_type_name, target_table):
    conn = pyodbc.connect(connection_string)
    cursor = conn.cursor()
    # 准备TVP参数
    tvp = cursor.execute(f"DECLARE @tvp {table_type_name}; SELECT * FROM @tvp;").setinputsizes([(pyodbc.SQL_SS_TABLE, table_type_name)])
    tvp.setoutputsize(pyodbc.SQL_SS_TABLE)
    # 绑定数据
    cursor.execute(f"INSERT INTO {target_table} SELECT * FROM ?", (chunk,))
    conn.commit()
    cursor.close()
    conn.close()

# 调用示例
for i, chunk in enumerate(chunks, start=1):
    start_time = time.time()
    insert_with_tvp(connection_string, chunk, "dbo.YourTableType", "dbo.YourTargetTable")
    end_time = time.time()
    print(f"{i * len(chunk)} records inserted in {end_time - start_time:.2f} seconds")

3. 临时禁用索引与约束 #

插入大量数据前，禁用非聚集索引、外键约束和触发器，插入完成后重新启用/重建：

-- 禁用非聚集索引
ALTER INDEX ALL ON dbo.YourTargetTable DISABLE;
-- 禁用外键约束
ALTER TABLE dbo.YourTargetTable NOCHECK CONSTRAINT ALL;

-- 插入数据后
-- 重建非聚集索引（比启用更快，能消除碎片）
ALTER INDEX ALL ON dbo.YourTargetTable REBUILD;
-- 启用外键约束
ALTER TABLE dbo.YourTargetTable CHECK CONSTRAINT ALL;

注意：操作前需确保没有其他并发写入，避免数据一致性问题

4. 调整连接与执行参数 #

• 在连接后执行SET NOCOUNT ON;：减少服务器返回的行数统计消息，降低网络开销
• 增加命令超时时间：设置cursor.timeout = 300（单位秒，根据块大小调整），或在连接字符串中加入Command Timeout=300
• 调整ODBC驱动参数：连接字符串中加入TDS_Version=7.4（适配SQL Server 2016+，提升兼容性和性能）

能否在每个块后重连？ #

可以，但会导致性能下降——每次建立TCP连接需要完成握手、认证等步骤，额外增加网络开销，尤其是远程服务器场景，会显著拉长总耗时。

更优的连接错误解决方案 #

优先排查并解决连接断开的根源，而非每块重连：

1. 增加超时设置：

   • 连接字符串中设置Connect Timeout=60（连接超时）和Command Timeout=300（命令执行超时）
   • 代码中设置cursor.timeout = 300

2. 启用连接池：
   pyodbc默认启用连接池，可通过连接字符串参数优化：

   connection_string = "DRIVER={ODBC Driver 17 for SQL Server};SERVER=your_server;DATABASE=your_db;UID=user;PWD=pwd;Pooling=True;Max Pool Size=10;Min Pool Size=2"
   

   连接池会复用现有连接，避免频繁创建新连接的开销。

3. 错误捕获与自动重试：
   仅在遇到连接错误时重连并重试当前块，而非每块都重连：

   import pyodbc
   import time
   
   def get_connection(connection_string):
       return pyodbc.connect(connection_string)
   
   connection_string = "your_connection_string"
   insert_query = "your_insert_query"
   chunk_size = 10000
   
   for i, chunk in enumerate(chunks, start=1):
       retries = 3
       success = False
       while retries > 0 and not success:
           try:
               conn = get_connection(connection_string)
               cursor = conn.cursor()
               cursor.fast_executemany = True
               start_time = time.time()
               cursor.executemany(insert_query, chunk)
               conn.commit()
               end_time = time.time()
               print(f"{i * chunk_size} records inserted in {end_time - start_time:.2f} seconds")
               cursor.close()
               conn.close()
               success = True
           except pyodbc.OperationalError as e:
               retries -= 1
               print(f"Connection error occurred: {str(e)}, retrying {retries} times...")
               time.sleep(5)  # 等待几秒后重试
               if conn:
                   try:
                       conn.close()
                   except:
                       pass
       if not success:
           print(f"Failed to insert chunk {i} after 3 retries")
   

SQLAlchemy底层依赖pyodbc作为驱动，其性能表现取决于使用方式：

• ORM层：如果用add_all或bulk_insert_mappings，会有ORM的对象映射开销，速度远不如直接用pyodbc的fast_executemany或TVP
• Core层：用SQLAlchemy Core的execute方法执行批量插入，性能接近pyodbc，但仍略逊于原生pyodbc（因为Core有额外的SQL构建和参数处理逻辑）

如果追求极致导入速度，直接使用pyodbc的原生优化（fast_executemany、TVP）是最优选择；如果需要ORM的便利性（比如数据模型映射、跨数据库兼容），SQLAlchemy是可行的，但需要接受一定的性能损耗。

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