使用 mssql-python 进行批量复制

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

如果你在同一会话中创建或修改目标表，请在 conn.commit() 之前调用 bulkcopy()。 批量复制协议使用独立的内部通道读取表元数据，因此未提交的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

DataFrame 采用列式结构，因此最快的方式是 bulkcopy_arrow()，它接收 pandas 已经能够生成的 Arrow 表。 bulkcopy()需要行元组，所以你需要先把列映射成 Python 对象。

在加载之前，先将 Arrow 表转换为目标列的数据类型。 pyarrow 为数值列推断出 float64，但驱动程序无法将其映射为 money、decimal 或 numeric：

import pandas as pd
import pyarrow as pa
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()

target = pa.schema([
    pa.field('ID', pa.int32()),
    pa.field('Name', pa.string()),
    pa.field('Amount', pa.decimal128(19, 4)),   # MONEY
])

table = pa.Table.from_pandas(df, preserve_index=False).cast(target)
result = cursor.bulkcopy_arrow("##PandasDemo", table)

如果不进行强制转换，加载将失败，并显示 ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money)。 使用 Table.cast() 构建类型转换，而不是将 schema 传递给 Table.from_pandas()，因为 Table.from_pandas() 无法直接将浮点列转换为 decimal128。 NaN 值在这条路径上会变成 SQL NULL ，所以你不需要先替换它们。

如果你改为需要行元组路径，那么当你传入 name=None 时，itertuples() 已经会返回元组：

data = list(df.itertuples(index=False, name=None))
result = cursor.bulkcopy("##PandasDemo", data)

加载 Apache Arrow 数据

使用 cursor.bulkcopy_arrow() 加载 Apache Arrow 数据。 这个方法直接从 Arrow 内存读取，所以调用前不会先构建 Python 行元组。

该 source 参数接受 pyarrow.Table、pyarrow.RecordBatch、pyarrow.RecordBatchReader 中的一个，或任何提供 Arrow C 数据接口的对象。 其余论证与 bulkcopy()相同。

import mssql_python
import pyarrow as pa

conn = mssql_python.connect(connection_string)

# bulkcopy_arrow() opens its own connection, so commit the table creation first.
conn.autocommit = True
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##ArrowDemo (ID INT, Name NVARCHAR(50), Amount FLOAT)
""")

table = pa.table({
    "ID": pa.array([1, 2, 3], type=pa.int32()),
    "Name": pa.array(["Alice", "Bob", "Carol"], type=pa.string()),
    "Amount": pa.array([50000.0, 60000.0, 55000.0], type=pa.float64()),
})

result = cursor.bulkcopy_arrow("##ArrowDemo", table)
print(f"Copied {result['rows_copied']} rows")

每个Arrow列类型必须与其目标SQL列类型兼容。 写入器不支持不同类型族之间的转换，因此将 float64 列传递给 money 列时，会在写入任何行之前引发 ValueError。 将decimal128用于货币、十进制和数值列。

传递箭源到 bulkcopy() ，会升 TypeError 起并引导你前往 bulkcopy_arrow()。

关于 Arrow 支持的更多信息，包括如何将一个表的结果集流向另一个表，请参见 Apache Arrow 集成。

处理 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	操作超时时间（秒） 适用于批量复制操作本身，不适用于内部连接。 使用 0 禁用操作超时。
keep_identity	False	保留源数据的身份值。
check_constraints	False	在加载期间检查表约束。
table_lock	False	获取表级锁而不是行级锁。
keep_nulls	False	保留 NULL 值，而不是插入列默认值。
fire_triggers	False	在目标表上触发 INSERT 触发器。
use_internal_transaction	False	将每个批次置于内部事务中处理。

Note

bulkcopy() 会与服务器单独建立一条内部连接。 该内部连接继承了光标的查询超时：在创建光标前设 Connection.timeout 为正值，且该值限制了批量复制连接尝试。 如果光标的查询超时为 0，内部连接使用默认的15秒连接超时。 游标在创建时会获取当时的值，因此之后更改 Connection.timeout 不会影响现有游标或正在进行中的批量复制操作。 在为慢速、受限流或高延迟端点（例如通过 VPN 或跨区域访问的端点）创建游标之前，请先调长查询超时时间。

处理错误

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()会调用一个行元组的迭代，因此每个值在复制开始前都必须作为 Python 对象存在。 当数据已经是列式时， bulkcopy_arrow() 直接读取Arrow缓冲区并跳过这一步。 pandas 或 Polars DataFrame、Parquet 文件以及 cursor.arrow() 的结果，都是 Arrow 源。 更多信息，请参见 加载 Apache Arrow 数据。

使用生成器处理大型数据集

生成器最小化内存占用，因为 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_arrow()	已经采用列式格式的大型数据集。	最快
cursor.bulkcopy()	来自面向行的数据源的大型数据集（超过 1,000 行）。	快速
cursor.executemany()	含参数的中等规模数据集。	温和
循环中的 cursor.execute()	小数据集，逻辑简单。	最慢

相关内容

• 用 mssql-python 执行查询
• mssql-python 的数据类型映射
• 使用mssql-python进行事务管理
• mssql-python 应用的性能调优