Kies een datalaad- en bewegingspatroon met mssql-python

De mssql-python driver biedt meerdere paden om data in Microsoft SQL te schrijven. Elk pad past bij verschillende werklasten. Deze gids helpt je de juiste te kiezen op basis van je datavolume, bronformaat en update-semantiek.

Bepaal op basis van werklast

Werklast Aanbevolen pad Waarom
Laad CSV-bestanden in een tabel CSV-gegevens laden met bulksgewijs kopiëren bulkcopy() met een generator verwerkt bestanden van elke grootte zonder ze in het geheugen te laden.
Voeg een enkele rij in uit applicatiecode Invoegingen van één rij Lage overhead, eenvoudige foutafhandeling, werkt met OUTPUT voor het teruggeven van gegenereerde sleutels.
Voeg een kleine tot matige batch toe uit applicatiecode Batchgewijze invoegingen Vermindert het aantal heen-en-weerbewegingen vergeleken met afzonderlijke inserts.
Laad honderden rijen of meer van elke bron bulksgewijs kopiëren TDS bulk-insert is de meest efficiënte methode voor grote volumes.
Rijen invoegen of bijwerken op basis van een sleutel Upsert met MERGE MERGE verwerkt INSERT, UPDATE, en DELETE in één stelling.
Laad een DataFrame in een tabel DataFrames laden Haal rijen uit pandas of Polars en voer deze in in bulkcopy().
Gegevens voorbereiden via Parquet-bestanden Parquet staging Nuttig voor cross-system ETL waarbij een tussentijds bestandsformaat nodig is.

Laad CSV-gegevens door deze bulksgewijs te kopiëren

Het laden van CSV-gegevens is de meest voorkomende ingest-vraag bij Python-databasewerk. Gebruik csv.reader met een generator die bulkcopy() voedt:

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

Het generatorpatroon houdt het geheugengebruik constant ongeacht de bestandsgrootte. Voor kolommapping en identiteitsbehandeling, zie Bulk copy operations.

Invoegingen van één rij

Gebruik single inserts voor applicatieniveau schrijfopdrachten waarbij je één record tegelijk verwerkt. Gebruik OUTPUT INSERTED om gegenereerde sleutels op te halen:

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

Afzonderlijke inzetstukken zijn de beste keuze als:

  • Je voegt per gebruikersactie één rij in (formulierindiening, API-aanroep).
  • Je moet elke rij afzonderlijk valideren of transformeren voordat je het invoegt.
  • Je hebt de ingevoegde ID of andere gegenereerde waarden direct nodig.

Invoegingen in batches

Gebruik executemany() wanneer je een matig aantal rijen hebt en de doorvoersnelheid van bulkkopiëren niet nodig hebt:

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() stuurt elke rij als een aparte geparametriseerde instructie. Wanneer doorvoersnelheid belangrijker is dan controle per rij, bulkcopy() is het efficiënter omdat het het TDS bulk insert-protocol gebruikt. Het omslagpunt hangt af van de rijbreedte en netwerklatentie, maar ligt doorgaans bij enkele honderden rijen.

Bulksgewijs kopiëren

Wanneer doorvoer belangrijker is dan controle per rij, gebruik bulkcopy(). Het gebruikt het TDS bulk insert-protocol, dat aanzienlijk efficiënter is dan row-by-row inserts:

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

Tips voor betere prestaties bij bulkkopiëren

  • Gebruik generatoren voor grote datasets om het geheugengebruik constant te houden.
  • Set batch_size om te bepalen hoeveel rijen per TDS-batch worden verzonden. Begin met 5.000 en pas aan op basis van de rijbreedte.
  • Gebruik tabelvergrendelingen voor exclusieve ladingen: cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True).
  • Schakel indexen uit voor het laden, en bouw daarna opnieuw op. Deze reeks voorkomt de indexonderhoudsoverhead tijdens de belasting.

Voor kolommappingen, identiteitskolommen, NULL-afhandeling en parallel laden, zie Bulk copy operations.

Upsert met MERGE

MERGEis de Microsoft SQL-instructie voor conditionele INSERT, UPDATE, en DELETE in één enkele operatie. Het behandelt het "insert if new, update if exists"-patroon dat Python-ontwikkelaars vaak nodig hebben.

Enkelrijige upsert

Voor één rij gebruik je MERGE met een USING-clausule die aliassen voor parameters definieert:

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 met een opsteltafel

Voor bulk-upserts plaats je de gegevens eerst in een tijdelijke tabel en gebruik je vervolgens MERGE om die gegevens van daaruit bij te werken. Gebruik insert-or-update als standaardpatroon voor DataFrame-upserts en batchupdates:

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

Dit voorbeeld toont het standaard invoeg- of updatepatroon:

  • INSERT rijen uit de bron die niet in het doel bestaan (WHEN NOT MATCHED BY TARGET).
  • UPDATE rijen die in beide voorkomen (WHEN MATCHED).
  • OUTPUT-clausule geeft aan welke actie er op elke rij is ondernomen, wat nuttig is voor auditsporen.

Waarschuwing

Voeg alleen toe WHEN NOT MATCHED BY SOURCE THEN DELETE wanneer de stagingdata een gezaghebbende volledige snapshot van het doel is. Als de batch alleen gewijzigde rijen bevat, verwijdert die clausule rijen die opzettelijk uit de bronfeed zijn weggelaten.

Als je volledige reconciliatie nodig hebt, breid MERGE dan pas uit nadat je hebt bevestigd dat de bron de gezaghebbende bron is voor de doeltabel:

WHEN NOT MATCHED BY SOURCE THEN
    DELETE

In gedeelde omgevingen gebruik per run een unieke globale tijdelijke tabelnaam of een permanente stagingtabel om botsingen tussen gelijktijdige taken te voorkomen.

Wanneer gebruik je in plaats daarvan afzonderlijke UPDATE- en INSERT-instructies

MERGE is krachtig, maar kent uitzonderingssituaties. Overweeg het gebruik van afzonderlijke statements wanneer:

  • Je hebt geen logica nodig DELETE . Een apart UPDATE gevolg door INSERT WHERE NOT EXISTS is leesbaarder en eenvoudiger te debuggen.
  • De MERGE uitspraak is complex genoeg dat het vergrendelingsgedrag moeilijk te voorspellen is. Aparte statements geven je expliciete controle over de vergrendelingsgranulariteit.
  • Je bent een tabel met hoge gelijktijdigheid aan het bijwerken waarbij MERGE lock-escalatie blokkades kan veroorzaken.
# 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()

DataFrames inladen

Haal rijen uit een pandas- of Polars DataFrame en laad ze met behulp van bulkcopy():

pandas

Converteer een pandas DataFrame naar tuples en geef door naar 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

Zet een Polars DataFrame om naar tuples met de methode .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()

Voor volledige laadpatronen van DataFrame, zie pandas-integratie en Polars-integratie.

Parquet-staging

Gebruik Parquet als tussenformaat bij het migreren van data tussen systemen of wanneer je ETL-pijplijn al Parquet-bestanden produceert:

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

Voor grote Parquet-bestanden, lees in rijgroepen om het geheugengebruik constant te houden:

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

Valideer geladen gegevens

Controleer na het laden het aantal rijen en controleer de gegevens steekproefs:

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

Vertrouw voor productieworkloads niet op de transactie van de aanroepende verbinding om een bulkcopy()-aanroep te beveiligen. bulkcopy() opent zijn eigen interne verbinding en legt de gekopieerde rijen onafhankelijk vast, dus een conn.rollback() op je primaire verbinding kan ze niet ongedaan maken. Twee benaderingen geven je atomiciteit:

  • Stel use_internal_transaction=True in dat elke batch in een eigen transactie wordt verpakt. Een batch die halverwege mislukt, wordt teruggedraaid in plaats van gedeeltelijk geladen achter te blijven.
  • Om gegevens te valideren voordat je die doorzet, kopieer je de gegevens in bulk naar een stagingtabel, valideer je ze en verplaats je de rijen vervolgens naar de doeltabel door binnen een transactie via je hoofdverbinding een INSERT ... SELECT te gebruiken. Omdat INSERT via je verbinding wordt uitgevoerd, maakt conn.rollback() dit ongedaan als de validatie mislukt.
# 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

Verwante onderwerpen

  • Last updated on