Verwenden Sie Bulk Copy mit mssql-python

Der mssql-python-Treiber enthält eine Bulk-Copy-Funktion, die große Datenmengen effizient in SQL Server, Azure SQL-Datenbank, Azure SQL Managed Instance und SQL-Datenbank in Microsoft Fabric einfügt.

Die Methode cursor.bulkcopy() bietet einen Hochleistungspfad zum Laden großer Datensätze:

• Minimiert Netzwerk-Roundtrips.
• Optional wird die Überprüfung von Constraints beim Laden umgangen.
• Verwendet das optimierte TDS-Bulk-Insert-Protokoll.
• Erreicht einen Durchsatz, der mit bcp.exe und SqlBulkCopyvergleichbar ist.

Die auf Rust basierende mssql_py_core native Erweiterung betreibt die Massenkopiefunktion. Es läuft außerhalb der normalen Cursor-execute()-Pipeline.

Grundlegende Nutzung

Rufen Sie bulkcopy() für einen Cursor auf und übergeben Sie dabei den Namen der Zieltabelle sowie eine iterierbare Folge von Zeilentupeln oder Row-Objekten:

import mssql_python

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

# Create a temp table for the demo
cursor.execute("""
    CREATE TABLE ##BulkDemo (
        ID INT,
        Name NVARCHAR(50),
        Amount MONEY
    )
""")
conn.commit()

data = [
    (1, "Alice", 50000.00),
    (2, "Bob", 60000.00),
    (3, "Carol", 55000.00),
]

result = cursor.bulkcopy("##BulkDemo", data)
print(f"Copied {result['rows_copied']} rows in {result['batch_count']} batch(es)")
print(f"Elapsed: {result['elapsed_time']}")

Zurückgegebener Wert

bulkcopy() gibt ein Wörterbuch zurück:

Methodensignatur

cursor.bulkcopy(
    table_name,                    # str – target table (can include schema, e.g. "dbo.MyTable")
    data,                          # Iterable[Tuple | Row] – rows to insert
    batch_size=0,                  # int – rows per batch; 0 = server optimal
    timeout=30,                    # int – operation timeout in seconds
    column_mappings=None,          # List[str] | List[Tuple[int,str]] | None
    keep_identity=False,           # bool – preserve identity values from source
    check_constraints=False,       # bool – check constraints during load
    table_lock=False,              # bool – use table-level lock
    keep_nulls=False,              # bool – preserve NULLs instead of defaults
    fire_triggers=False,           # bool – fire INSERT triggers on target
    use_internal_transaction=False, # bool – use internal transaction per batch
)

Spaltenzuordnungen

Standardmäßig ordnet bulkcopy() Spalten nach ihrer Ordinalposition zu. Jede Datenspalte entspricht der Tabellenspalte mit demselben Index. Nutze den Parameter, column_mappings um dieses Verhalten zu überschreiben.

Spaltennamenliste

Jede Position in der Liste entspricht dem Quelldatenindex:

result = cursor.bulkcopy(
    "##BulkDemo",
    data,
    column_mappings=["ID", "Name", "Amount"],
)

Fortgeschrittenes Format: explizite Indexabbildung

Jedes Tupel nimmt die Form (source_index, target_column_name)an. Verwenden Sie dieses Format, um Spalten zu überspringen oder neu zu ordnen:

result = cursor.bulkcopy(
    "##BulkDemo",
    data,
    column_mappings=[(0, "ID"), (1, "Name"), (2, "Amount")],
)

Laden aus Dateien

Du kannst Daten aus CSV-Dateien und anderen Dateiformaten laden, indem du einen Generator an bulkcopy()übergibst.

CSV-Datei

import csv
import io
import mssql_python

# In production, replace io.StringIO with open("data.csv", "r", ...)
csv_data = """ID,Name,Value
1,Widget,9.99
2,Gadget,24.50
3,Gizmo,4.75
"""

def csv_row_generator(file_obj):
    """Generator that yields tuples from a CSV file object."""
    reader = csv.reader(file_obj)
    next(reader)  # Skip header
    for row in reader:
        if row:  # skip blank lines
            yield (
                int(row[0]),      # ID
                row[1],           # Name
                float(row[2]),    # Value
            )

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

cursor.execute("""
    CREATE TABLE ##CSVImport (ID INT, Name NVARCHAR(100), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy("##CSVImport", csv_row_generator(io.StringIO(csv_data)))
print(f"Imported {result['rows_copied']} rows from CSV")

Große Dateien mit Batching

Stellen Sie den Parameter batch_size so ein, dass er steuert, wie viele Zeilen der Treiber pro Batch sendet. Dieser Ansatz funktioniert gut für große Dateien:

import csv
import io
import mssql_python

# In production, replace io.StringIO with open("large_file.csv", "r", ...)
csv_data = "\n".join(
    ["ID,Name,Value"] + [f"{i},Item {i},{i * 1.5}" for i in range(1, 201)]
)

def csv_rows(file_obj):
    reader = csv.reader(file_obj)
    next(reader)  # Skip header
    for row in reader:
        if row:
            yield (int(row[0]), row[1], float(row[2]))

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

cursor.execute("""
    CREATE TABLE ##LargeCSV (ID INT, Name NVARCHAR(100), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy(
    "##LargeCSV",
    csv_rows(io.StringIO(csv_data)),
    batch_size=50,
)
print(f"Imported {result['rows_copied']} rows in {result['batch_count']} batches")

Laden Sie pandas DataFrames

Wandle einen pandas DataFrame in eine Liste von Tupeln um, bevor du ihn an bulkcopy() übergibst:

import pandas as pd
import mssql_python

df = pd.DataFrame({
    'ID': [1, 2, 3],
    'Name': ['Alice', 'Bob', 'Carol'],
    'Amount': [50000.0, 60000.0, 55000.0],
})

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

cursor.execute("""
    CREATE TABLE ##PandasDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()

data = [tuple(row) for row in df.itertuples(index=False, name=None)]
result = cursor.bulkcopy("##PandasDemo", data)

Handle NULL-Werte

Übergeben Sie None an einer beliebigen Spaltenposition, um einen SQL-NULL-Wert einzufügen:

cursor.execute("""
    CREATE TABLE ##NullDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()

data = [
    (1, "Alice", 50000.00),
    (2, "Bob", None),       # NULL Amount
    (3, None, 55000.00),    # NULL Name
]

cursor.bulkcopy("##NullDemo", data)

Identitätsspalten

Um explizite Identitätswerte einzufügen, setze keep_identity=True:

cursor.execute("""
    CREATE TABLE ##IdentDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()

data = [
    (100, "Alice", 50000.00),
    (200, "Bob", 60000.00),
]

cursor.bulkcopy("##IdentDemo", data, keep_identity=True)

Wenn keep_identity=False (die Standardeinstellung) verwendet wird, lassen Sie die Identitätsspalte in Ihren Daten weg und verwenden Sie column_mappings, um die Spalten ohne Identitätseigenschaft anzusprechen.

Massenkopieoptionen

Fehler behandeln

bulkcopy() erzeugt eine Ausnahme, wenn die Last fehlschlägt, also wickelt man den Aufruf in einen try/except Block, um Fehler zu erkennen. Beachte, dass bulkcopy() über eine eigene interne Verbindung ausgeführt wird und die kopierten Zeilen unabhängig festschreibt, sodass ein conn.rollback() auf deiner Hauptverbindung sie nicht rückgängig machen kann. Um einen Batch atomar auszuführen, setzen Sie use_internal_transaction=True, wodurch jeder Batch in eine eigene Transaktion eingeschlossen wird, die automatisch zurückgesetzt wird, wenn der Batch fehlschlägt:

import mssql_python

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

cursor.execute("""
    CREATE TABLE ##ImportDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
conn.commit()

data = [
    (1, "Alice", 50000.00),
    (2, "Bob", 60000.00),
    (3, "Carol", 55000.00),
]

try:
    result = cursor.bulkcopy("##ImportDemo", data, use_internal_transaction=True)
    print(f"Successfully copied {result['rows_copied']} rows")
except (mssql_python.DatabaseError, ValueError) as e:
    # bulkcopy() commits on its own connection, so there's nothing to roll back
    # here. With use_internal_transaction=True, a failed batch is already rolled
    # back on the bulk copy connection.
    print(f"Bulk copy failed: {e}")

Um einen Ladevorgang mit Ihrer eigenen Validierungslogik abzusichern, kopieren Sie die Daten zunächst per Masseneinfügung in eine Staging-Tabelle und überführen Sie die Zeilen anschließend mit einem INSERT ... SELECT innerhalb einer Transaktion über Ihre Hauptverbindung in die Zieltabelle. Das INSERT wird über deine Verbindung ausgeführt, daher macht conn.rollback() es rückgängig, wenn die Validierung fehlschlägt.

Authentifizierung

Massenkopien verwenden einen separaten internen Kanal, der ein eigenes Token benötigt. Der Treiber übernimmt automatisch die Tokenerfassung für die unterstützten Authentifizierungsmethoden.

Verwaltete Identität (ActiveDirectoryMSI)

Verwendung Authentication=ActiveDirectoryMSI für systemzugewiesene oder benutzerdefinierte verwaltete Identität. Diese Authentifizierungsmethode wird für Azure-gehostete Dienste wie Azure-VMs, App Service, Functions und AKS empfohlen.

import mssql_python

# System-assigned managed identity
conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryMSI;"
    "Encrypt=yes"
)
cursor = conn.cursor()

cursor.execute("CREATE TABLE ##MsiDemo (ID INT, Name NVARCHAR(50))")
conn.commit()

result = cursor.bulkcopy("##MsiDemo", [(1, "Alice"), (2, "Bob")])
print(f"Copied {result['rows_copied']} rows")

Für eine benutzerzugewiesene verwaltete Identität geben Sie die Client-ID in der Verbindungszeichenfolge weiter:

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryMSI;"
    "UID=<client-id>;"
    "Encrypt=yes"
)

Dienstprinzipal (ActiveDirectoryServicePrincipal)

Verwenden Sie Authentication=ActiveDirectoryServicePrincipal für die Service Principal-Authentifizierung (Clientanmeldeinformationen).

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryServicePrincipal;"
    "UID=<application-client-id>;"
    "PWD=<client-secret>;"
    "Encrypt=yes"
)
cursor = conn.cursor()

cursor.execute("CREATE TABLE ##SpDemo (ID INT, Value FLOAT)")
conn.commit()

result = cursor.bulkcopy("##SpDemo", [(1, 1.5), (2, 2.5)])
print(f"Copied {result['rows_copied']} rows")

Standard-Zugangsdatenkette (ActiveDirectoryDefault)

ActiveDirectoryDefault Versucht mehrere Zugangsdatenanbieter nacheinander, wie Umgebungsvariablen, Workload-Identität, verwaltete Identität und mehr. Es funktioniert sowohl für lokale Entwicklung als auch für Azure-gehostete Dienste ohne Codeänderungen.

Weitere Informationen zur Authentifizierung finden Sie unter Microsoft Entra-Authentifizierung.

Leistungstipps

Die folgenden Techniken helfen Ihnen, den Durchsatz beim Massenkopieren zu maximieren.

Verwenden Sie Generatoren für große Datensätze

Generatoren minimieren den Speicherverbrauch, da bulkcopy() jedes iterierbare Objekt akzeptiert:

def data_generator(count):
    """Generate rows without loading all into memory."""
    for i in range(count):
        yield (i, f"Item {i}", i * 1.5)

cursor = conn.cursor()
cursor.execute("""
    CREATE TABLE ##LargeDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy("##LargeDemo", data_generator(1000))

Verwenden Sie Tabellensperren für schnellere Ladevorgänge

Wenn Sie keine gleichzeitigen Leser haben, stellen Sie ein, table_lock=True um den Sperraufwand bei großen Anfangslasten zu reduzieren.

result = cursor.bulkcopy(
    "##LargeDemo",
    data,
    table_lock=True,
    batch_size=100000,
)

Deaktiviere Indizes während des Ladens

Deaktivieren Sie vorübergehend nicht-geclusterte Indizes vor der Massenbelastung und bauen Sie sie danach wieder auf, um die Leistung zu verbessern:

cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##IndexDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
cursor.execute("CREATE NONCLUSTERED INDEX IX_Name ON ##IndexDemo(Name)")
conn.commit()

cursor.execute("ALTER INDEX IX_Name ON ##IndexDemo DISABLE")
conn.commit()

result = cursor.bulkcopy("##IndexDemo", data)
conn.commit()

cursor.execute("ALTER INDEX IX_Name ON ##IndexDemo REBUILD")
conn.commit()

Tabellen parallel laden

Eröffne für jede Tabelle eine separate Verbindung und führe die Lasten gleichzeitig aus.

import concurrent.futures

def load_table(table_name, rows):
    conn = mssql_python.connect(connection_string)
    cursor = conn.cursor()
    cursor.execute(f"CREATE TABLE {table_name} (ID INT, Name NVARCHAR(50), Value FLOAT)")
    conn.commit()
    result = cursor.bulkcopy(table_name, rows)
    conn.commit()
    conn.close()
    return result["rows_copied"]

data = [(i, f"Item {i}", i * 1.5) for i in range(100)]

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = [
        executor.submit(load_table, "##Load1", data),
        executor.submit(load_table, "##Load2", data),
        executor.submit(load_table, "##Load3", data),
    ]
    for future in concurrent.futures.as_completed(futures):
        print(f"Loaded {future.result()} rows")

Vergleich mit Alternativen

Die folgende Tabelle vergleicht Massenkopien mit anderen Dateneinfügungsmethoden.

Verwandte Inhalte

• Abfragen ausführen
• Datentypzuordnungen
• Transaktionsverwaltung
• Leistungsoptimierung