Välj ett dataladdnings- och rörelsemönster med mssql-python

Drivrutinen mssql-python erbjuder flera vägar för att skriva data till Microsoft SQL. Varje väg passar olika arbetsbelastningar. Denna guide hjälper dig att välja rätt baserat på din datavolym, källformat och uppdateringssemantik.

Bestäm efter arbetsbelastning

Arbetsbörda Rekommenderad sökväg Varför
Ladda CSV-filer i en tabell Ladda CSV-data med bulkkopiering bulkcopy() med en generator hanterar filer av vilken storlek som helst utan att ladda in dem i minnet.
Sätt in en enda rad från applikationskoden Infogning av enstaka rader Låg overhead, enkel felhantering, fungerar med OUTPUT för att returnera genererade nycklar.
Infoga en liten eller medelstor batch från programkod Batchade insatser Minskar antalet tur- och retur-anrop jämfört med enskilda infogningar.
Ladda hundratals rader eller fler från vilken källa som helst Masskopiering TDS-massinfogning är den mest effektiva metoden för stora volymer.
Infoga eller uppdatera rader baserat på en nyckel Upsert med MERGE MERGE hanterar INSERT, UPDATE, och DELETE i en sats.
Ladda en DataFrame i en tabell Ladda DataFrames Ta ut rader från pandor eller polarare och mata till bulkcopy().
Mellanlagra data med Parquet-filer Parkettuppsättning Användbart för tvärsystem-ETL där ett mellanliggande filformat behövs.

Ladda CSV-data med bulkkopiering

Att ladda CSV-data är den vanligaste inläsningsfrågan för Python-databasarbete. Använd csv.reader med en generator som matar bulkcopy():

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()

Generatormönstret håller minnesanvändningen konstant oavsett filstorlek. Information om kolumnmappning och hantering av identitetskolumner finns i Masskopieringsåtgärder.

Enkelradinfogningar

Använd enskilda infogningar för skrivningar på applikationsnivå när du bearbetar en post i taget. Användning OUTPUT INSERTED för att hämta genererade nycklar:

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()

Enskilda insatser är rätt val när:

  • Du infogar en rad per användarhandling (formulärinlämning, API-anrop).
  • Du behöver validera eller transformera varje rad individuellt innan du infogar.
  • Du behöver det insatta ID:t eller andra genererade värden omedelbart.

Batchade insatser

Använd executemany() när du har ett måttligt antal rader och inte behöver den höga dataöverföringshastighet som bulkkopiering ger:

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() skickar varje rad som en separat parameteriserad sats. När genomströmning är viktigare än kontroll på radnivå är bulkcopy() mer effektivt eftersom det använder TDS-protokollet för massinfogning. Brytpunkten beror på radbredden och nätverkslatensen, men den ligger vanligtvis på några hundra rader.

Masskopiering

När genomströmningen är viktigare än kontroll per rad, använd bulkcopy(). Den använder TDS bulkinsert-protokoll, som är betydligt effektivare än rad-för-rad-insättningar:

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()

Prestandatips för masskopiering

  • Använd generatorer för stora datamängder för att hålla minnesanvändningen konstant.
  • Set batch_size för att kontrollera hur många rader som skickas per TDS-batch. Börja med 5 000 och justera efter radbredd.
  • Använd table locks för exklusiva laddningar: cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True).
  • Inaktivera index innan laddning, bygg sedan om efteråt. Denna sekvens undviker underhållskostnaderna för index under inläsningen.

För kolumnmappningar, identitetskolumner, NULL-hantering och parallell laddning, se Bulkkopieringsoperationer.

Upsert med MERGE

MERGEär Microsoft SQL:s sats för villkorlig INSERT, UPDATE, och DELETE i en enda operation. Den hanterar mönstret "infoga om ny, uppdatera om det finns"-mönstret som Python-utvecklare ofta behöver.

Enkelradigt uppsert

För en enda rad, använd MERGE med en USING klausul som definierar parameteralias:

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()

Bulk-upsert med en uppställningstabell

För bulk-upserts, läs först in data i en temporär tabell och använd sedan MERGE för att uppdatera från den. Använd insert-or-update som standardmönster för DataFrame-upserts och batchuppdateringar:

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()

Detta exempel visar standardmönstret för insättning eller uppdatering:

  • INSERT rader från källan som inte finns i målet (WHEN NOT MATCHED BY TARGET).
  • UPDATE rader som finns i båda (WHEN MATCHED).
  • OUTPUT-klausulen rapporterar vilken åtgärd som vidtogs på varje rad, vilket är användbart för revisionsspår.

Caution

Lägg endast till WHEN NOT MATCHED BY SOURCE THEN DELETE när stagingdata utgör en fullständig och tillförlitlig ögonblicksbild av målet. Om batchen endast innehåller ändrade rader, raderar den klausulen rader som avsiktligt utelämnades från källmaten.

Om du behöver fullständig avstämning, utöka endast MERGE efter att du bekräftat att källan är auktoritativ för måltabellen:

WHEN NOT MATCHED BY SOURCE THEN
    DELETE

I delade miljöer bör du använda ett unikt namn på en global temporär tabell per körning eller en permanent mellantabell för att undvika konflikter mellan jobb som körs samtidigt.

När ska separata UPDATE- och INSERT-satser användas istället

MERGE är kraftfull men har undantagsfall. Överväg att använda separata påståenden när:

  • Du behöver ingen DELETElogik. En separat UPDATE följt av INSERT WHERE NOT EXISTS är mer läsbar och enkel att felsöka.
  • Påståendet MERGE är tillräckligt komplext för att låsbeteendet ska vara svårt att förutsäga. Separata instruktioner ger dig uttrycklig kontroll över låsnivån.
  • Du uppdaterar en tabell med hög samtidighet där MERGE-låseskalering kan orsaka blockering.
# 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()

Ladda DataFrames

Extrahera rader från en pandas- eller Polars DataFrame genom att använda bulkcopy():

pandas

Konvertera en pandas DataFrame till tupler och skicka till 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

Konvertera en Polars DataFrame till tupler med metoden .rows() :

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()

För fullständiga DataFrame-laddningsmönster, se pandas-integration och Polars-integration.

Parkettuppsättning

Använd Parquet som ett mellanformat när du migrerar data mellan system eller när din ETL-pipeline redan producerar Parquet-filer:

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()

För stora Parquet-filer, läs i radgrupper för att hålla minnesanvändningen konstant:

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()

Validera inladdad data

När data har lästs in, kontrollera antal rader och gör stickprovskontroller av data:

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}")

Vid produktionsbelastning ska du inte förlita dig på den anropande anslutningens transaktion för att skydda ett anrop till bulkcopy(). bulkcopy() öppnar sin egen interna anslutning och bekräftar de kopierade raderna oberoende, så en conn.rollback() i huvudanslutningen kan inte ångra dem. Två metoder ger dig atomicitet:

  • Ställ use_internal_transaction=True in att varje batch ska wrappas i en egen transaktion. En batch som misslyckas delvis rullar tillbaka den batchen istället för att lämna den halvlastad.
  • För att validera data innan du för över dem masskopierar du dem till en mellanlagringstabell, validerar dem och flyttar sedan raderna till måltabellen med hjälp av en INSERT ... SELECT i en transaktion via huvudanslutningen. Eftersom detta INSERT körs via din anslutning, conn.rollback() återställer det om valideringen misslyckas.
# 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

Relaterat innehåll

  • Last updated on