使用 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

如果你在同一会话中创建或修改目标表，请在 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() 在循环中 小数据集，逻辑简单。 最慢

相关内容

  • Last updated on