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	bulkcopy_arrow()läser DataFrames Arrow-data utan att bygga ett Python-objekt för varje värde.
Ladda in Apache Arrow-data i en tabell	Ladda Arrow-data	bulkcopy_arrow()läser Arrow-minnet direkt, utan att bygga Python-tupler.
Mellanlagra data med Parquet-filer	Parkettuppsättning	Användbart för tvärsystem-ETL där ett mellanliggande filformat behövs. Parquet är redan data i Arrow-format, så det laddas utan konvertering av rader.

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 bulk insert-protokollet, som strömmar rader istället för att skicka ett uttalande per rad:

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.
• Användning bulkcopy_arrow() när källkoden är kolumnformad, såsom en DataFrame eller en Parquet-fil. Den hoppar över konverteringen till Python-radtupler.
• 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

En DataFrame är kolumnformad, så ladda den med bulkcopy_arrow() istället för att platta ut den till radtupler för bulkcopy().

Matcha piltyperna med destinationskolumnerna innan du laddar. pyarrow härleder float64 för en numerisk kolumn, som drivrutinen inte kan mappa till money, decimal eller numeric.

pandas

import pandas as pd
import pyarrow as pa

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

target = pa.schema([
    pa.field("Name", pa.string()),
    pa.field("ProductNumber", pa.string()),
    pa.field("ListPrice", pa.decimal128(19, 4)),   # MONEY
])

table = pa.Table.from_pandas(
    df[["Name", "ProductNumber", "ListPrice"]], preserve_index=False
).cast(target)

cursor.bulkcopy_arrow("dbo.ProductImport", table)
conn.commit()

Använd Table.cast() istället för att skicka schemat till Table.from_pandas(), vilket inte kan konvertera en flyttalkolumn till decimal128 direkt.

Polars

Polars implementerar Arrow C-datagränssnittet, så du kan skicka själva DataFrame-objektet. Gjuta kolonnerna först av samma anledning:

import polars as pl

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

cursor.bulkcopy_arrow(
    "dbo.ProductImport",
    df.select([
        "Name",
        "ProductNumber",
        pl.col("ListPrice").cast(pl.Decimal(19, 4)),   # MONEY
    ]),
)
conn.commit()

Du kan också ställa in typerna när du läser filen, med pl.read_csv("products.csv", schema_overrides={"ListPrice": pl.Decimal(19, 4)}).

Att skicka DataFrame-objektet direkt överlämnar dess buffertar till drivrutinen utan att kopieras. df.to_arrow() fungerar också, men Polars omkodar strängkolumner under den konverteringen, vilket kopierar all strängdata.

bulkcopy_arrow() tar emot en pyarrow.Table, en RecordBatch, en RecordBatchReader eller valfritt objekt som implementerar Arrow C-data-gränssnittet via __arrow_c_stream__ eller __arrow_c_array__. Att skicka någon av dessa till bulkcopy() utlöser TypeError.

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

Ladda pildata

När källkoden redan är i Apache Arrow-format, cursor.bulkcopy_arrow() laddar den utan att bygga Python-tupler först.

from decimal import Decimal

import pyarrow as pa

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

table = pa.table({
    "Name": pa.array(["Widget", "Gadget"], type=pa.string()),
    "ProductNumber": pa.array(["WI-1000", "GA-2000"], type=pa.string()),
    "ListPrice": pa.array([Decimal("29.99"), Decimal("49.99")], type=pa.decimal128(10, 2)),
})

result = cursor.bulkcopy_arrow("dbo.ProductImport", table, batch_size=5000)
print(f"Copied {result['rows_copied']} rows")

Metoden accepterar också en pyarrow.RecordBatch eller en pyarrow.RecordBatchReader, så att du kan strömma en resultatuppsättning direkt från cursor.arrow_reader() till en annan tabell.

Varje Arrow-kolumntyp måste vara kompatibel med sin destinations-SQL-kolumntyp, och skrivaren konverterar inte mellan typfamiljer. För mer information, se Apache Arrow-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. En Parquet-fil läser in Arrow-data, så skicka den direkt till bulkcopy_arrow():

import pyarrow.parquet as pq

cursor.bulkcopy_arrow("dbo.ProductImport", pq.read_table("products.parquet"))
conn.commit()

För stora Parquet-filer, iterera radgrupper för att hålla minnesanvändningen konstant. Varje sats är en RecordBatch, som bulkcopy_arrow() accepterar direkt:

import pyarrow.parquet as pq

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

for batch in parquet_file.iter_batches(batch_size=10000):
    cursor.bulkcopy_arrow("dbo.ProductImport", batch)

conn.commit()

För att streama hela filen i ett enda anrop, omslut batcherna med en RecordBatchReader:

import pyarrow as pa
import pyarrow.parquet as pq

parquet_file = pq.ParquetFile("products.parquet")
reader = pa.RecordBatchReader.from_batches(
    parquet_file.schema_arrow, parquet_file.iter_batches(batch_size=10000)
)

cursor.bulkcopy_arrow("dbo.ProductImport", reader)
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

• Använd bulk-kopiering med mssql-python
• Använd mssql-python med pandas
• Använd mssql-python med Polars
• Prestandaoptimering för mssql-python-applikationer
• Transaktionshantering med mssql-python
• Välj ett dataåtkomst- och analysmönster med mssql-python