mssql-python 驅動程式包含批量複製功能，能有效將大量資料插入 SQL Server、Azure SQL Database、Azure SQL 受控執行個體 及 Microsoft Fabric 中的 SQL 資料庫。
此 cursor.bulkcopy() 方法提供了載入大型資料集的高效能路徑：
- 減少網路往返次數。
- 可選擇性地在載入時繞過限制檢查。
- 使用最佳化的 TDS 大量插入通訊協定。
- 可達成與
bcp.exeSqlBulkCopy相當的吞吐量。
基於 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 資料框
將 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 VMS、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"
)
Service principal （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() |
大型資料集（超過 1,000 筆資料列）。 | 最快 |
cursor.executemany() |
含參數的中型資料集 | Moderate |
cursor.execute() 在迴圈中 |
小資料集，邏輯簡單明瞭。 | 最慢 |