Wybierz wzór ładowania i przemieszczania danych za pomocą mssql-python

Sterownik mssql-python oferuje wiele ścieżek zapisu danych do Microsoft SQL. Każda ścieżka nadaje się do różnych obciążeń roboczych. Ten przewodnik pomaga wybrać odpowiedni model na podstawie ilości danych, formatu źródła i semantyki aktualizacji.

Decyduj według obciążenia

Obciążenie Zalecana ścieżka Dlaczego
Załaduj pliki CSV do tabeli Załaduj dane CSV za pomocą kopiowania zbiorczego bulkcopy() Generator obsługuje pliki dowolnego rozmiaru bez ładowania ich do pamięci.
Wstaw pojedynczy wiersz z kodu aplikacji Wstawianie pojedynczego wiersza Niewielki narzut, prosta obsługa błędów, działa z OUTPUT przy zwracaniu wygenerowanych kluczy.
Wstawianie niewielkiej lub średniej partii danych za pomocą kodu aplikacji Wkładki batched Zmniejsza to liczbę podróży w obie strony w porównaniu do pojedynczych insertów.
Załaduj setki wierszy lub więcej z dowolnego źródła Kopiowanie masowe Wkładka TDS bulk to najwydajniejsza droga dla dużych objętości.
Wstaw lub aktualizuj wiersze na podstawie klucza Upsert za pomocą MERGE MERGE obsługuje INSERT, UPDATE, oraz DELETE w jednym zdaniu.
Załaduj DataFrame do tabeli Ładowanie ramek danych Wyodrębnij rzędy z pand lub polarnych i podaj do bulkcopy().
Przygotuj dane za pomocą plików Parquet Scenografia parkietowa Przydatne dla ETL międzysystemowego, gdzie potrzebny jest format pliku pośredni.

Załaduj dane CSV za pomocą kopiowania zbiorczego

Ładowanie danych CSV to najczęściej używane pytanie dotyczące pobierania danych w bazie Python. Użyj csv.reader z generatorem zasilającym 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()

Wzorzec generatora utrzymuje stałe zużycie pamięci niezależnie od rozmiaru pliku. W przypadku mapowania kolumn i obsługi tożsamości zobacz operacje kopiowania masowego.

Wkładki jednowierszowe

Używaj pojedynczych insertów do zapisów na poziomie aplikacji, gdzie przetwarzasz jeden rekord na raz. Zastosowanie OUTPUT INSERTED do odzyskania wygenerowanych kluczy:

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

Pojedyncze wkładki są właściwym wyborem, gdy:

  • Na każdą akcję użytkownika (przesłanie formularza, wywołanie API) wstawiasz jeden wiersz.
  • Musisz zweryfikować lub przekształcić każdy wiersz osobno przed wstawieniem.
  • Potrzebujesz natychmiast wstawionego identyfikatora lub innych wygenerowanych wartości.

Wstawianie wsadowe

Używaj executemany(), gdy masz średnią liczbę wierszy i nie potrzebujesz wysokiej wydajności kopiowania zbiorczego:

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() wysyła każdy wiersz jako osobne, parametryzowane zdanie. Gdy przepustowość ma większe znaczenie niż kontrola poszczególnych wierszy, bulkcopy() jest wydajniejsza, ponieważ wykorzystuje protokół zbiorczego wstawiania TDS. Crossover zależy od szerokości wiersza i opóźnienia sieci, ale zazwyczaj występuje w niskich setkach wierszy.

Kopiowanie zbiorcze

Gdy przepustowość ma większe znaczenie niż kontrola na wiersz, używaj bulkcopy(). Wykorzystuje protokół TDS bulk insert, który jest znacznie bardziej wydajny niż wstawianie wiersz po wierszu:

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

Wskazówki dotyczące wydajności przy kopiowaniu zbiorczym

  • Używaj generatorów do dużych zbiorów danych, aby utrzymać stałe zużycie pamięci.
  • Ustaw batch_size, aby kontrolować, ile wierszy jest wysyłanych w każdej partii TDS. Zacznij od 5 000 i dostosuj w zależności od szerokości wiersza.
  • Używaj blokad tabeli do ładowania wyłącznego: cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True).
  • Wyłącz indeksy przed ładowaniem, a potem buduj ponownie. Ta sekwencja pozwala uniknąć narzutu związanego z utrzymywaniem indeksu podczas ładowania.

W przypadku mapowania kolumn, kolumn tożsamościowych, obsługi NULL i ładowania równoległego, zobacz operacje kopiowania masowego.

Upsert za pomocą MERGE

MERGE to instrukcja Microsoft SQL do warunkowego INSERT, UPDATE i DELETE w ramach jednej operacji. Obsługuje wzorzec "wstaw, jeśli nowy, zaktualizuj, jeśli istnieje", którego programiści Python często potrzebują.

Upsert z pojedynczym rzędem

Dla pojedynczego wiersza użyj MERGE z klauzulą USING, która definiuje aliasy parametrów:

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

Zwiększenie objętości z tablicą etapową

W przypadku zbiorczych operacji upsert najpierw umieść dane w tabeli tymczasowej, a następnie użyj MERGE, aby zaktualizować dane na podstawie tej tabeli. Stosuj insert-or-update jako domyślny wzorzec dla upsertów DataFrame i aktualizacji wsadowych:

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

Ten przykład pokazuje domyślny wzorzec wstawienia lub aktualizacji:

  • INSERT wiersze ze źródła, które nie istnieją w miejscu docelowym (WHEN NOT MATCHED BY TARGET).
  • UPDATE wiersze występujące w obu (WHEN MATCHED).
  • Klauzula OUTPUT raportuje, jakie działania zostały podjęte w każdym wierszu, co jest przydatne w śledzeniach audytu.

Caution

Dodawaj WHEN NOT MATCHED BY SOURCE THEN DELETE tylko wtedy, gdy dane przejściowe stanowią autorytatywną pełną migawkę systemu docelowego. Jeśli partia zawiera tylko zmienione wiersze, ta klauzula usuwa wiersze, które zostały celowo pominięte w źródłowym feedzie.

Jeśli potrzebujesz pełnego uzgodnienia, rozszerz MERGE dopiero po potwierdzeniu, że źródło jest miarodajne dla tabeli docelowej:

WHEN NOT MATCHED BY SOURCE THEN
    DELETE

W środowiskach współdzielonych używaj unikalnej globalnej nazwy tabeli tymczasowej dla każdego uruchomienia lub trwałej tabeli pośredniej, aby uniknąć kolizji między jednocześnie uruchamianymi zadaniami.

Kiedy zamiast tego używać oddzielnych instrukcji UPDATE i INSERT

MERGE jest potężny, ale ma przypadki graniczne. Rozważ użycie oddzielnych zdań, gdy:

  • Nie potrzebujesz DELETE logiki. Oddzielne UPDATE, po którym następuje INSERT WHERE NOT EXISTS, jest bardziej czytelne i łatwiejsze do debugowania.
  • Instrukcja MERGE jest na tyle złożona, że trudno przewidzieć działanie mechanizmu blokowania. Oddzielne instrukcje dają wyraźną kontrolę nad szczegółowością blokady.
  • Aktualizujesz tabelę o dużej współbieżności, w której eskalacja blokad MERGE może powodować blokowanie.
# 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()

Załaduj ramki danych

Wyodrębnij wiersze z dataFrame pandas lub Polars i załaduj je za pomocą bulkcopy():

pandas

Przekonwertuj obiekt DataFrame biblioteki pandas do krotek i przekaż go do 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

Konwertowanie ramki danych Polars do krotek za pomocą metody .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()

Aby poznać pełne wzorce ładowania DataFrame, zobacz integrację z pandas oraz integrację z Polars.

Scenografia parkietowa

Używaj Parquet jako formatu pośredniego podczas migracji danych między systemami lub gdy Twój potok ETL już generuje pliki 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()

W przypadku dużych plików Parquet odczytuj dane w grupach wierszy, aby zachować stałe zużycie pamięci:

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

Weryfikacja załadowanych danych

Po załadowaniu sprawdź liczbę wierszy i wybiórczo sprawdź dane:

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

W przypadku obciążeń produkcyjnych nie polegaj na transakcji połączenia wywołującego w celu ochrony połączenia bulkcopy() . bulkcopy() otwiera własne wewnętrzne połączenie i niezależnie zatwierdza skopiowane wiersze, więc operacja conn.rollback() na głównym połączeniu nie może ich cofnąć. Dwa podejścia dają atomowość:

  • Ustaw use_internal_transaction=True, aby każda partia była objęta osobną transakcją. Partia, w której podczas przetwarzania wystąpi błąd, jest wycofywana w całości, zamiast pozostawać częściowo załadowana.
  • Aby zweryfikować dane przed ich przeniesieniem, skopiuj je zbiorczo do tabeli przejściowej, zweryfikuj je, a następnie przenieś wiersze do tabeli docelowej, używając elementu INSERT ... SELECT w ramach transakcji na głównym połączeniu. Ponieważ to INSERT działa w Twoim połączeniu, conn.rollback() cofa to, jeśli walidacja się nie powiedzie.
# 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

Treści powiązane

  • Last updated on