用 mssql-python 選擇資料載入與移動模式

驅動程式mssql-python提供多條路徑將資料寫入 Microsoft SQL。每條路徑適合不同的工作量。本指南將根據你的資料量、來源格式及更新語意，幫助你選擇合適的方案。

依照工作量決定

[工作負載]	建議的路徑	原因為何
將 CSV 檔案載入表格	將 CSV 資料與批量複製一起載入	`bulkcopy()` 有了產生器，可以處理任意大小的檔案，而不必將檔案載入記憶體。
從應用程式程式碼插入一個資料列	單列插入	低開銷、簡單易懂的錯誤處理，搭配 OUTPUT 來回傳產生的金鑰。
從應用程式碼插入小到中等數量的批次	批次插入	相較於單筆插入，可減少往返次數。
從任何來源載入數百列甚至更多	大量複製	TDS 大量插入是處理大量資料最有效率的方式。
根據鍵值插入或更新資料列	插入或更新 MERGE	`MERGE` 處理 INSERT、 UPDATE和 DELETE 在一個陳述中。
將資料框架載入資料表	載入資料幀	從 pandas 或 Polars 擷取資料列，並傳送給 `bulkcopy()`。
透過Parquet檔案取得階段資料	拼花舞台	對於需要中間檔案格式的跨系統 ETL 非常有用。

將 CSV 資料與批量複製一起載入

載入 CSV 資料是 Python 資料庫工作中最常見的導入問題。在發電機為 bulkcopy() 供電時使用 csv.reader：

import csv
import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

# Create a target table
cursor.execute("""
    IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'ProductImport')
    CREATE TABLE dbo.ProductImport (
        Name nvarchar(100),
        ProductNumber nvarchar(25),
        ListPrice decimal(10,2)
    )
""")
conn.commit()

def csv_rows(path):
    with open(path, newline="", encoding="utf-8") as f:
        reader = csv.reader(f)
        next(reader)  # Skip header
        for row in reader:
            yield (row[0], row[1], float(row[2]))

result = cursor.bulkcopy(
    "dbo.ProductImport",
    csv_rows("products.csv"),
    batch_size=5000
)
print(f"Loaded {result['rows_copied']} rows")
conn.commit()

產生器模式無論檔案大小如何，都能保持記憶體使用不變。關於欄位映射與身份處理，請參見批量複製操作。

單列插入

當您一次處理一筆記錄時，對應用程式層級的寫入請使用單筆插入。用於 OUTPUT INSERTED 擷取已產生的金鑰：

cursor.execute("""
    INSERT INTO dbo.ProductImport (Name, ProductNumber, ListPrice)
    OUTPUT INSERTED.Name
    VALUES (%(name)s, %(product_number)s, %(list_price)s)
""", {"name": "Widget", "product_number": "WG-1000", "list_price": 19.99})

inserted_name = cursor.fetchval()
conn.commit()

單片式嵌件在以下情況下是理想的選擇：

你會為每個使用者動作（提交表單、API 呼叫）插入一列。
你需要在插入前，先對每一列個別進行驗證或轉換。
你需要立即取得插入後的 ID 或其他產生的值。

批次插入

當資料列數量適中，且不需要大量複製的吞吐量時，請使用 executemany()：

rows = [
    {"name": "Widget A", "product_number": "WG-1001", "list_price": 19.99},
    {"name": "Widget B", "product_number": "WG-1002", "list_price": 24.99},
    {"name": "Widget C", "product_number": "WG-1003", "list_price": 29.99},
]

cursor.executemany(
    "INSERT INTO dbo.ProductImport (Name, ProductNumber, ListPrice) VALUES (%(name)s, %(product_number)s, %(list_price)s)",
    rows
)
conn.commit()

executemany() 將每列作為獨立的參數化語句傳送。當吞吐量比每列控制更重要時， bulkcopy() 它更有效率，因為它使用 TDS 批量插入協定。交叉點取決於列寬度和網路延遲，但通常只有幾百列左右。

大量複製

當吞吐量比每列控制更重要時，請使用 bulkcopy()。它使用 TDS 大量插入協定，其效率明顯優於逐列插入：

rows = [
    ("Widget A", "WG-1001", 19.99),
    ("Widget B", "WG-1002", 24.99),
    ("Widget C", "WG-1003", 29.99),
]

result = cursor.bulkcopy("dbo.ProductImport", rows, batch_size=5000)
print(f"Loaded {result['rows_copied']} rows")
conn.commit()

大量複製的效能提示

使用產生器 處理大型資料集，以保持記憶體使用量不變。
設定 batch_size，以控制每個 TDS 批次傳送的資料列數。從 5,000 開始，根據行寬調整。
使用桌鎖 來處理排他性載入： cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True)。
載入前先停用索引，載入後再重建。此序列避免了負載期間的指標維護負擔。

關於欄位映射、身份欄位、NULL 處理及平行載入，請參見批量複製操作。

使用 MERGE 進行更新插入

MERGE是 Microsoft SQL 中用於條件式 INSERT、UPDATE 和 DELETE 的陳述式，並可在單一作業中完成。它可處理 Python 開發人員常見的「若為新資料則插入，若已存在則更新」這種模式。

單一資料列更新或插入

對於單一列，使用 MERGE 搭配可定義參數別名的 USING 子句：

cursor.execute("""
    MERGE dbo.ProductImport AS target
    USING (SELECT %(name)s AS Name, %(product_number)s AS ProductNumber, %(list_price)s AS ListPrice) AS source
    ON target.ProductNumber = source.ProductNumber
    WHEN MATCHED THEN
        UPDATE SET
            Name = source.Name,
            ListPrice = source.ListPrice
    WHEN NOT MATCHED THEN
        INSERT (Name, ProductNumber, ListPrice)
        VALUES (source.Name, source.ProductNumber, source.ListPrice);
""", {"name": "Widget A", "product_number": "WG-1001", "list_price": 24.99})
conn.commit()

使用暫存資料表進行大量 upsert

對於大量更新，先把資料分階段放進暫存表，再用 MERGE 來更新。使用 insert-or-update 作為 DataFrame 插入或更新作業與批次更新的預設模式：

import csv
import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

# Step 1: Create a global temp table for staging
# Note: bulkcopy() requires global temp tables (##), not session temp tables (#)
cursor.execute("""
    IF OBJECT_ID('tempdb..##ProductImportStage') IS NOT NULL
        DROP TABLE ##ProductImportStage;
    CREATE TABLE ##ProductImportStage (
        Name nvarchar(100),
        ProductNumber nvarchar(25),
        ListPrice decimal(10,2)
    )
""")
cursor.commit()

# Step 2: Bulk load into the staging table
def csv_rows(path):
    with open(path, newline="", encoding="utf-8") as f:
        reader = csv.reader(f)
        next(reader)
        for row in reader:
            yield (row[0], row[1], float(row[2]))

cursor.bulkcopy("##ProductImportStage", csv_rows("products_update.csv"), batch_size=5000)

# Step 3: MERGE from staging into the target table
cursor.execute("""
    MERGE dbo.ProductImport AS target
    USING ##ProductImportStage AS source
    ON target.ProductNumber = source.ProductNumber
    WHEN MATCHED THEN
        UPDATE SET
            Name = source.Name,
            ListPrice = source.ListPrice
    WHEN NOT MATCHED BY TARGET THEN
        INSERT (Name, ProductNumber, ListPrice)
        VALUES (source.Name, source.ProductNumber, source.ListPrice)
    OUTPUT $action, INSERTED.ProductNumber, DELETED.ProductNumber;
""")

# Step 4: Read the OUTPUT to see what changed
for row in cursor.fetchall():
    print(f"{row[0]}: inserted={row[1]}, deleted={row[2]}")

conn.commit()

此範例展示了預設的插入或更新模式：

INSERT 來自來源但不存在於目標中的資料列（WHEN NOT MATCHED BY TARGET）。
UPDATE 同時存在於兩個（WHEN MATCHED）中的行。
OUTPUT 子句報告每列所採取的行動，對稽核軌跡非常有用。

Caution

僅當暫存資料是目標的具權威性的完整快照時，才新增 WHEN NOT MATCHED BY SOURCE THEN DELETE。如果該批次只包含已變更的資料列，則該子句會刪除來源饋送中刻意省略的資料列。

如果你需要完整對帳，請在確認來源對目標表具有權威性後，再延長：MERGE

WHEN NOT MATCHED BY SOURCE THEN
    DELETE

在共享環境中，請於每次執行時使用唯一的全域暫存資料表名稱，或使用永久暫存資料表，以避免並行作業之間發生衝突。

何時應改用個別的 UPDATE 和 INSERT 陳述式

MERGE 很強大，但也有例外。考慮在下列情況下使用個別陳述式：

你不需要 DELETE 邏輯。將獨立的 UPDATE 放在前面，後面接著 INSERT WHERE NOT EXISTS，可讀性更高，也更容易除錯。
這個 MERGE 陳述本身就很複雜，鎖定行為很難預測。獨立的陳述句讓你能明確控制鎖的細節。
你正在更新一個高並行存取的資料表，而在此情況下，MERGE 鎖定升級可能會導致阻塞。

# Simpler alternative: UPDATE then INSERT
cursor.execute("""
    UPDATE dbo.ProductImport
    SET Name = %(name)s, ListPrice = %(list_price)s
    WHERE ProductNumber = %(product_number)s
""", {"name": "Widget A", "list_price": 24.99, "product_number": "WG-1001"})

if cursor.rowcount == 0:
    cursor.execute("""
        INSERT INTO dbo.ProductImport (Name, ProductNumber, ListPrice)
        VALUES (%(name)s, %(product_number)s, %(list_price)s)
    """, {"name": "Widget A", "product_number": "WG-1001", "list_price": 24.99})

conn.commit()

載入資料幀

從 pandas 或 Polars DataFrame 擷取資料列，並使用 bulkcopy() 載入：

pandas

將 pandas DataFrame 轉換為元組，並傳遞給 bulkcopy()：

import pandas as pd

df = pd.read_csv("products.csv")

# Convert DataFrame rows to tuples
rows = list(df[["Name", "ProductNumber", "ListPrice"]].itertuples(index=False, name=None))

cursor.bulkcopy("dbo.ProductImport", rows, batch_size=5000)
conn.commit()

極地 (Polars)

使用 .rows() 方法將 Polars DataFrame 轉換為元組：

import polars as pl

df = pl.read_csv("products.csv")

# Convert Polars DataFrame to list of tuples
rows = df.select(["Name", "ProductNumber", "ListPrice"]).rows()

cursor.bulkcopy("dbo.ProductImport", rows, batch_size=5000)
conn.commit()

完整的資料幀載入模式，請參見 pandas 積分與 Polars 積分。

拼花舞台

在系統間遷移資料或當您的 ETL 管線已經產生 Parquet 檔案時，請使用 Parquet 作為中間格式：

import pyarrow.parquet as pq

# Read Parquet file
table = pq.read_table("products.parquet")

# Convert to rows for bulkcopy
rows = [tuple(row) for row in zip(*[col.to_pylist() for col in table.columns])]

cursor.bulkcopy("dbo.ProductImport", rows, batch_size=5000)
conn.commit()

對於大型 Parquet 檔案，請以列群組讀取以保持記憶體使用量不變：

import pyarrow.parquet as pq

parquet_file = pq.ParquetFile("products.parquet")

for batch in parquet_file.iter_batches(batch_size=10000):
    rows = [tuple(row) for row in zip(*[col.to_pylist() for col in batch.columns])]
    cursor.bulkcopy("dbo.ProductImport", rows, batch_size=10000)

conn.commit()

驗證已載入的資料

載入後，請驗證資料列數並抽樣檢查資料：

cursor.execute("SELECT COUNT(*) FROM dbo.ProductImport")
count = cursor.fetchval()
print(f"Total rows: {count}")

cursor.execute("""
    SELECT TOP 5 Name, ProductNumber, ListPrice
    FROM dbo.ProductImport
    ORDER BY Name
""")
for row in cursor:
    print(f"  {row.Name} ({row.ProductNumber}): ${row.ListPrice:.2f}")

對於生產負載，不要依賴呼叫連線的交易來保護通話 bulkcopy() 。 bulkcopy() 會開啟自己的內部連線，並獨立認可複製的資料列，因此主連線上的 conn.rollback() 無法復原這些資料列。有兩種方法能賦予原子性：

設定 use_internal_transaction=True 為將每個批次包裝成獨立的交易。若某批次在執行到一半時失敗，系統會回滾該批次，而不會讓它處於只載入一半的狀態。
若要在將資料提升至正式環境前先驗證，請先大量複製到暫存資料表，完成驗證後，再在主連線的交易中使用 INSERT ... SELECT 將資料列移至目標資料表。因為該 INSERT 是在你的連線上執行，若驗證失敗，conn.rollback() 就會撤銷該操作。

# Stage the data. bulkcopy() runs on its own connection, so these rows
# persist regardless of the transaction below.
cursor.bulkcopy("dbo.ProductImport_Stage", rows, batch_size=5000)

try:
    cursor.execute("SELECT COUNT(*) FROM dbo.ProductImport_Stage")
    count = cursor.fetchval()

    if count < expected_count:
        raise ValueError(f"Expected {expected_count} rows, got {count}")

    # This INSERT runs on your connection, so it's covered by the transaction.
    cursor.execute("""
        INSERT INTO dbo.ProductImport (Name, ProductNumber, ListPrice)
        SELECT Name, ProductNumber, ListPrice FROM dbo.ProductImport_Stage
    """)
    conn.commit()
except Exception:
    conn.rollback()
    raise

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-22