mssql-python驱动包含批量复制功能，能够高效地将大量数据插入SQL Server、Azure SQL 数据库、Azure SQL 托管实例和Microsoft Fabric中的SQL数据库。
该 cursor.bulkcopy() 方法为加载大型数据集提供了高效路径：
- 减少网络往返次数。
- 可选择性地绕过加载时的约束检查。
- 采用优化的TDS批量插入协议。
- 实现了可与
bcp.exe和SqlBulkCopy相媲美的吞吐量。
基于 Rust mssql_py_core 的原生扩展支持批量复制功能。 它在正常的光标 execute() 流水线之外运行。
基本用法
在光标上调用 bulkcopy() ，传递目标表名称和一串行元组或 Row 对象的迭代：
Important
如果你在同一会话中创建或修改目标表，请在 bulkcopy() 之前调用 conn.commit()。 批量复制协议使用独立的内部通道读取表元数据，因此未提交的DDL更改可能导致死锁或超时。
import mssql_python
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
# Create a temp table for the demo
cursor.execute("""
CREATE TABLE ##BulkDemo (
ID INT,
Name NVARCHAR(50),
Amount MONEY
)
""")
conn.commit()
data = [
(1, "Alice", 50000.00),
(2, "Bob", 60000.00),
(3, "Carol", 55000.00),
]
result = cursor.bulkcopy("##BulkDemo", data)
print(f"Copied {result['rows_copied']} rows in {result['batch_count']} batch(es)")
print(f"Elapsed: {result['elapsed_time']}")
返回值
bulkcopy() 返回一个词典：
| Key | 类型 | 描述 |
|---|---|---|
rows_copied |
int | 成功复制的行数。 |
batch_count |
int | 处理批次数量。 |
elapsed_time |
float | 操作所需时间（秒）。 |
方法签名
cursor.bulkcopy(
table_name, # str – target table (can include schema, e.g. "dbo.MyTable")
data, # Iterable[Tuple | Row] – rows to insert
batch_size=0, # int – rows per batch; 0 = server optimal
timeout=30, # int – operation timeout in seconds
column_mappings=None, # List[str] | List[Tuple[int,str]] | None
keep_identity=False, # bool – preserve identity values from source
check_constraints=False, # bool – check constraints during load
table_lock=False, # bool – use table-level lock
keep_nulls=False, # bool – preserve NULLs instead of defaults
fire_triggers=False, # bool – fire INSERT triggers on target
use_internal_transaction=False, # bool – use internal transaction per batch
)
列映射
默认情况下，bulkcopy() 按序号位置映射列。 每个数据列都对应于表中相同索引位置的列。 使用参数 column_mappings 来覆盖这个行为。
栏名列表
列表中的每个位置对应源数据索引：
result = cursor.bulkcopy(
"##BulkDemo",
data,
column_mappings=["ID", "Name", "Amount"],
)
高级格式：显式索引映射
每个元组的形式为 (source_index, target_column_name)。 使用此格式跳过或重新排序列：
result = cursor.bulkcopy(
"##BulkDemo",
data,
column_mappings=[(0, "ID"), (1, "Name"), (2, "Amount")],
)
从文件加载
你可以通过将生成器传递给 bulkcopy()，从 CSV 文件和其他文件格式中加载数据。
CSV 文件
import csv
import io
import mssql_python
# In production, replace io.StringIO with open("data.csv", "r", ...)
csv_data = """ID,Name,Value
1,Widget,9.99
2,Gadget,24.50
3,Gizmo,4.75
"""
def csv_row_generator(file_obj):
"""Generator that yields tuples from a CSV file object."""
reader = csv.reader(file_obj)
next(reader) # Skip header
for row in reader:
if row: # skip blank lines
yield (
int(row[0]), # ID
row[1], # Name
float(row[2]), # Value
)
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE ##CSVImport (ID INT, Name NVARCHAR(100), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy("##CSVImport", csv_row_generator(io.StringIO(csv_data)))
print(f"Imported {result['rows_copied']} rows from CSV")
带批处理的大文件
设置 batch_size 参数控制驱动每批发送多少行。 这种方法适用于大型文件：
import csv
import io
import mssql_python
# In production, replace io.StringIO with open("large_file.csv", "r", ...)
csv_data = "\n".join(
["ID,Name,Value"] + [f"{i},Item {i},{i * 1.5}" for i in range(1, 201)]
)
def csv_rows(file_obj):
reader = csv.reader(file_obj)
next(reader) # Skip header
for row in reader:
if row:
yield (int(row[0]), row[1], float(row[2]))
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE ##LargeCSV (ID INT, Name NVARCHAR(100), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy(
"##LargeCSV",
csv_rows(io.StringIO(csv_data)),
batch_size=50,
)
print(f"Imported {result['rows_copied']} rows in {result['batch_count']} batches")
加载 pandas DataFrames
将 pandas DataFrame 转换为元组列表，然后传递给 bulkcopy()：
import pandas as pd
import mssql_python
df = pd.DataFrame({
'ID': [1, 2, 3],
'Name': ['Alice', 'Bob', 'Carol'],
'Amount': [50000.0, 60000.0, 55000.0],
})
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE ##PandasDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()
data = [tuple(row) for row in df.itertuples(index=False, name=None)]
result = cursor.bulkcopy("##PandasDemo", data)
处理 NULL 值
在任意列位置传入 None 以插入 SQL NULL 值：
cursor.execute("""
CREATE TABLE ##NullDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()
data = [
(1, "Alice", 50000.00),
(2, "Bob", None), # NULL Amount
(3, None, 55000.00), # NULL Name
]
cursor.bulkcopy("##NullDemo", data)
标识列
若要插入显式的标识值，请将 keep_identity=True 设置为：
cursor.execute("""
CREATE TABLE ##IdentDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()
data = [
(100, "Alice", 50000.00),
(200, "Bob", 60000.00),
]
cursor.bulkcopy("##IdentDemo", data, keep_identity=True)
当 keep_identity=False（默认）时，请在数据中省略标识列，并使用 column_mappings 以针对非标识列。
批量复制选项
| 参数 | 默认 | 描述 |
|---|---|---|
batch_size |
0 |
每批行数。
0 让服务器选择最优大小。 |
timeout |
30 |
操作超时时间（秒） |
keep_identity |
False |
保留源数据的身份值。 |
check_constraints |
False |
加载时检查表约束。 |
table_lock |
False |
获取表级锁，而不是行级锁。 |
keep_nulls |
False |
保留 NULL 值，而不是插入列默认值。 |
fire_triggers |
False |
在目标表上触发 INSERT 触发器。 |
use_internal_transaction |
False |
将每个批次置于内部事务中处理。 |
处理错误
bulkcopy() 如果加载失败，会触发异常，因此将调用包裹在 try/except 块中以捕捉错误。 请记住，bulkcopy() 使用其自身的内部连接运行，并会独立提交复制过来的行，因此主连接上的 conn.rollback() 无法撤销这些行。 要使批处理原子化，设置 use_internal_transaction=True，将每个批次包裹在其独立事务中，若批处理失败会自动回滚：
import mssql_python
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE ##ImportDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
conn.commit()
data = [
(1, "Alice", 50000.00),
(2, "Bob", 60000.00),
(3, "Carol", 55000.00),
]
try:
result = cursor.bulkcopy("##ImportDemo", data, use_internal_transaction=True)
print(f"Successfully copied {result['rows_copied']} rows")
except (mssql_python.DatabaseError, ValueError) as e:
# bulkcopy() commits on its own connection, so there's nothing to roll back
# here. With use_internal_transaction=True, a failed batch is already rolled
# back on the bulk copy connection.
print(f"Bulk copy failed: {e}")
要让加载过程受您自己的验证逻辑控制，请先批量复制到暂存表中，然后在主连接的事务中通过 INSERT ... SELECT 将这些行转入目标表。 该操作 INSERT 在当前连接上运行，因此如果验证失败，conn.rollback() 会撤销该操作。
Authentication
批量复制使用一个独立的内部通道，需要独立的令牌。 驱动程序会自动处理支持的认证方法的令牌获取。
管理身份（ActiveDirectoryMSI）
对于系统分配的托管标识或用户分配的托管标识，请使用 Authentication=ActiveDirectoryMSI。 该认证方法推荐用于Azure托管服务，如Azure虚拟机、App Service、Functions和AKS。
import mssql_python
# System-assigned managed identity
conn = mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Authentication=ActiveDirectoryMSI;"
"Encrypt=yes"
)
cursor = conn.cursor()
cursor.execute("CREATE TABLE ##MsiDemo (ID INT, Name NVARCHAR(50))")
conn.commit()
result = cursor.bulkcopy("##MsiDemo", [(1, "Alice"), (2, "Bob")])
print(f"Copied {result['rows_copied']} rows")
对于用户指定的托管身份，通过连接字符串传递客户端ID：
conn = mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Authentication=ActiveDirectoryMSI;"
"UID=<client-id>;"
"Encrypt=yes"
)
服务主体（ActiveDirectoryServicePrincipal）
使用 Authentication=ActiveDirectoryServicePrincipal 进行服务主体（客户端凭证）身份验证。
conn = mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Authentication=ActiveDirectoryServicePrincipal;"
"UID=<application-client-id>;"
"PWD=<client-secret>;"
"Encrypt=yes"
)
cursor = conn.cursor()
cursor.execute("CREATE TABLE ##SpDemo (ID INT, Value FLOAT)")
conn.commit()
result = cursor.bulkcopy("##SpDemo", [(1, 1.5), (2, 2.5)])
print(f"Copied {result['rows_copied']} rows")
默认凭证链（ActiveDirectoryDefault）
ActiveDirectoryDefault 依次尝试多个凭证提供者，如环境变量、工作负载身份、托管身份等。 它既适用于本地开发，也能在无需修改代码的情况下运行于Azure托管服务。
有关认证的更多信息，请参见 Microsoft Entra 认证。
性能提示
以下技术帮助您最大化批量复制吞吐量。
使用生成器处理大型数据集
生成器最小化内存占用，因为 bulkcopy() 接受任意可迭代：
def data_generator(count):
"""Generate rows without loading all into memory."""
for i in range(count):
yield (i, f"Item {i}", i * 1.5)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE ##LargeDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy("##LargeDemo", data_generator(1000))
使用表锁以加快加载速度
在没有并发读取器的情况下，设置 table_lock=True 以减少大规模初始加载时的锁定开销。
result = cursor.bulkcopy(
"##LargeDemo",
data,
table_lock=True,
batch_size=100000,
)
加载时禁用索引
在批量加载前暂时禁用非集群索引，然后重构它们以提升性能：
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE ##IndexDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
cursor.execute("CREATE NONCLUSTERED INDEX IX_Name ON ##IndexDemo(Name)")
conn.commit()
cursor.execute("ALTER INDEX IX_Name ON ##IndexDemo DISABLE")
conn.commit()
result = cursor.bulkcopy("##IndexDemo", data)
conn.commit()
cursor.execute("ALTER INDEX IX_Name ON ##IndexDemo REBUILD")
conn.commit()
并行加载表
为每个表开启独立连接，同时运行加载。
import concurrent.futures
def load_table(table_name, rows):
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
cursor.execute(f"CREATE TABLE {table_name} (ID INT, Name NVARCHAR(50), Value FLOAT)")
conn.commit()
result = cursor.bulkcopy(table_name, rows)
conn.commit()
conn.close()
return result["rows_copied"]
data = [(i, f"Item {i}", i * 1.5) for i in range(100)]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = [
executor.submit(load_table, "##Load1", data),
executor.submit(load_table, "##Load2", data),
executor.submit(load_table, "##Load3", data),
]
for future in concurrent.futures.as_completed(futures):
print(f"Loaded {future.result()} rows")
与替代方案的比较
下表比较了批量复制与其他数据插入方法。
| 方法 | 用例 | 性能 |
|---|---|---|
cursor.bulkcopy() |
大型数据集（超过1000行）。 | 最快 |
cursor.executemany() |
含参数的中等规模数据集。 | 温和 |
cursor.execute() 在循环中 |
小数据集，逻辑简单。 | 最慢 |