如何通过Python高效向Azure SQL Server插入10万+行数据

针对你在Azure Function中用pyodbc批量插入10万行数据耗时4-5分钟的问题，以下是几个针对性的优化方案：

1. 开启pyodbc的fast_executemany参数 #

这是提升pyodbc批量插入性能最直接的手段，默认关闭，开启后会将参数化查询的批量操作转换为更高效的TDS批量请求，大幅减少网络交互次数。

示例代码：

import pyodbc

conn_str = (
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=your-server.database.windows.net;"
    "DATABASE=your-db;"
    "UID=your-username;"
    "PWD=your-password;"
    "Encrypt=yes;"
    "TrustServerCertificate=no;"
    "Connection Timeout=30;"
)

insert_sql = """
INSERT INTO your_table (col1, col2, col3)
VALUES (?, ?, ?)
"""

data = [("val1", "val2", "val3"), ...]  # 你的列表的列表数据

with pyodbc.connect(conn_str) as conn:
    cursor = conn.cursor()
    # 开启fast_executemany
    cursor.fast_executemany = True
    cursor.executemany(insert_sql, data)
    conn.commit()

2. 分批次插入 #

一次性插入10万行可能导致内存占用过高或数据库锁等待，建议拆分批次（比如每批5000-10000行），配合事务提交，平衡性能和稳定性。

示例代码：

batch_size = 5000
total_rows = len(data)

with pyodbc.connect(conn_str) as conn:
    cursor = conn.cursor()
    cursor.fast_executemany = True
    for i in range(0, total_rows, batch_size):
        batch = data[i:i+batch_size]
        cursor.executemany(insert_sql, batch)
        conn.commit()

3. 使用Azure Blob存储配合BULK INSERT #

如果数据量极大，可先将数据写入CSV文件上传到Azure Blob存储，再通过Azure SQL的BULK INSERT或OPENROWSET(BULK...)命令直接从Blob读取数据插入，这是性能最优的方案之一（适合10万+级数据）。

核心步骤：

• 将列表数据转换为CSV格式，保存到Azure Function临时目录
• 使用Azure Storage SDK将CSV上传到Blob存储（需配置SAS令牌或托管身份）
• 执行SQL批量插入命令：

BULK INSERT your_table
FROM 'https://your-storage.blob.core.windows.net/container/your-file.csv'
WITH (
    FORMAT = 'CSV',
    FIRSTROW = 2,  -- 若CSV包含表头
    FIELDTERMINATOR = ',',
    ROWTERMINATOR = '\n',
    CREDENTIAL = 'your-storage-credential'  -- 预先创建的数据库凭据
)

4. 调整连接与数据库配置 #

• 连接字符串中添加Max Pool Size=100（根据Azure Function并发量调整），避免频繁创建销毁连接
• 确保Azure SQL Server服务层级足够（如S3或更高），避免资源瓶颈
• 插入前临时关闭目标表的非主键索引/约束，插入完成后再重建，减少插入时的索引维护开销

5. 用pandas配合to_sql批量插入 #

如果已经在用pandas处理数据，可直接用to_sql方法，配合method='multi'或fast_executemany=True（pandas 1.4+支持）：

import pandas as pd

df = pd.DataFrame(data, columns=["col1", "col2", "col3"])
with pyodbc.connect(conn_str) as conn:
    df.to_sql(
        name="your_table",
        con=conn,
        if_exists="append",
        index=False,
        method="multi",
        chunksize=5000
    )

优先级建议 #

优先尝试开启fast_executemany，这个改动最小，通常能将插入时间压缩到几十秒级别；如果性能仍不满足，再考虑分批次或Blob配合BULK INSERT的方案。

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