Gebruik bulk copy met mssql-python

De mssql-python-driver bevat een bulkkopieerfunctie die efficiënt grote hoeveelheden data invoert in SQL Server, Azure SQL Database, Azure SQL Managed Instance en SQL database in Microsoft Fabric.

De cursor.bulkcopy() methode biedt een high-performance pad voor het laden van grote datasets:

• Minimaliseert netwerk-retourreizen.
• Omzeilt optioneel de controle van beperkingen tijdens het laden.
• Gebruikt het geoptimaliseerde TDS bulk insert-protocol.
• Bereikt een doorvoer vergelijkbaar met bcp.exe en SqlBulkCopy.

De op Rust gebaseerde mssql_py_core native extensie ondersteunt de bulkkopieerfunctie. Het werkt buiten de normale cursor-execute()pipeline.

Basaal gebruik

Roep bulkcopy() aan op een cursor en geef de naam van de doeltabel en een iterable van rijtuples of Row-objecten door:

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

Retourwaarde

bulkcopy() Geeft een woordenboek terug:

Methodesignatuur

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
)

Kolomtoewijzingen

Standaard brengt bulkcopy() kolommen in kaart op basis van hun positie. Elke datakolom wordt gekoppeld aan de tabelkolom in dezelfde index. Gebruik de column_mappings parameter om dit gedrag te overrulen.

Lijst van kolomnamen

Elke positie in de lijst komt overeen met de brondata-index:

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

Geavanceerd formaat: expliciete indexmapping

Elke tuple heeft de vorm (source_index, target_column_name). Gebruik dit formaat om kolommen over te slaan of te herschikken:

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

Laden vanuit bestanden

Je kunt gegevens laden uit CSV-bestanden en andere bestandsformaten door een generator aan bulkcopy()te geven.

CSV-bestand

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

Grote bestanden met batching

Stel de batch_size parameter in om te bepalen hoeveel rijen de driver per batch verzendt. Deze aanpak werkt goed voor grote bestanden:

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

Laad pandas DataFrames

Converteer een pandas DataFrame naar een lijst van tuples voordat je het doorgeeft aan bulkcopy():

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)

NULL-waarden verwerken

Geef None een willekeurige kolompositie door om een SQL-waarde NULL in te voegen:

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)

ID-kolommen

Om expliciete identiteitswaarden in te voegen, stelt u keep_identity=True in:

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)

Wanneer keep_identity=False is ingesteld (standaard), laat u de identiteitskolom weg uit uw gegevens en gebruikt u column_mappings om de niet-identiteitskolommen te selecteren.

Bulkkopieeropties

Afhandeling van fouten

bulkcopy() Genereert een uitzondering als de load faalt, dus wikkel de aanroep in een try/except blok om fouten te vangen. Houd er rekening mee dat het bulkcopy() op zijn eigen interne verbinding draait en de gekopieerde rijen onafhankelijk commit, dus een conn.rollback() op je hoofdverbinding kan ze niet ongedaan maken. Om een batch atomair te maken, stel use_internal_transaction=True, die elke batch in een eigen transactie wikkelt die automatisch terugrolt als de batch faalt:

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

Om het laden van gegevens te laten afhangen van je eigen validatielogica, kopieer je de gegevens eerst in bulk naar een stagingtabel en verplaats je de rijen vervolgens naar de doeltabel met een INSERT ... SELECT binnen een transactie via je hoofdverbinding. Dit INSERT draait via je verbinding, dus conn.rollback() maakt het ongedaan als de validatie mislukt.

Authentication

Bulkcopy gebruikt een apart intern kanaal dat een eigen token vereist. De driver verwerkt automatisch tokenverwerving voor de ondersteunde authenticatiemethoden.

Beheerde identiteit (ActiveDirectoryMSI)

Gebruik Authentication=ActiveDirectoryMSI voor systeem- of door de gebruiker toegewezen beheerde identiteit. Deze authenticatiemethode wordt aanbevolen voor door Azure gehoste diensten zoals Azure VM's, App Service, Functions en AKS.

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

Voor een door de gebruiker toegewezen beheerde identiteit geef je de client-ID door in de verbindingsreeks:

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

Service principal (ActiveDirectoryServicePrincipal)

Gebruik Authentication=ActiveDirectoryServicePrincipal voor service principal (clientgegevens) authenticatie.

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

Standaard inloggegevensketen (ActiveDirectoryDefault)

ActiveDirectoryDefault probeert meerdere referentieproviders opeenvolgend uit, zoals omgevingsvariabelen, workloadidentiteit, beheerde identiteit en meer. Het werkt zowel voor lokale ontwikkeling als voor Azure-gehoste diensten zonder codewijzigingen.

Voor meer informatie over authenticatie, zie Microsoft Entra authenticatie.

Tips voor prestaties

De volgende technieken helpen je om de bulk-kopieerdoorvoer te maximaliseren.

Gebruik generatoren voor grote datasets

Generatoren minimaliseren het geheugenverbruik omdat bulkcopy() elk itereerbaar object accepteert:

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

Gebruik tabelvergrendelingen voor snellere laadbewerkingen

Als je geen gelijktijdige lezers hebt, stel table_lock=True dan in om de vergrendelingsoverhead tijdens grote initiële ladingen te verminderen.

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

Schakel indexen uit tijdens het laden

Schakel niet-geclusterde indexen tijdelijk uit vóór de bulkbelasting en bouw ze daarna opnieuw op voor betere prestaties:

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

Laad tabellen parallel

Open voor elke tabel een aparte verbinding en voer de laadbewerkingen parallel uit.

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

Vergelijking met alternatieven

De volgende tabel vergelijkt bulkkopieën met andere methoden voor gegevensinvoeging.

Verwante onderwerpen

• Uitvoeren van queries
• Gegevenstypetoewijzingen
• Transactiebeheer
• Performance-optimalisatie