Använd bulk-kopiering med mssql-python

mssql-python-drivrutinen inkluderar en bulkkopieringsfunktion som effektivt infogar stora mängder data i SQL Server, Azure SQL Database, Azure SQL Managed Instance och SQL-databas i Microsoft Fabric.

Metoden cursor.bulkcopy() erbjuder en högpresterande väg för att ladda stora datamängder:

Minimerar nätverkets tur-och-retur-resor.
Undviker valfritt begränsningskontroll under belastning.
Använder det optimerade TDS bulk insert-protokollet.
Uppnår genomströmning jämförbar med bcp.exe och SqlBulkCopy.

Den Rust-baserade mssql_py_core inbyggda tillägget driver bulkkopieringsfunktionen. Den körs utanför det normala markör- execute() flödet.

Grundläggande användning

Anropa bulkcopy() på en markör och ange namnet på måltabellen samt en itererbar samling med radtupler eller Row-objekt:

Important

Om du skapar eller ändrar måltabellen i samma session, anropa conn.commit() före bulkcopy(). Bulkkopieringsprotokollet använder en separat intern kanal för att läsa tabellmetadata, så en obunden DDL-ändring kan orsaka deadlock eller timeout.

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

Returvärde

bulkcopy() återlämnar en ordbok:

Key	Type	Beskrivning
`rows_copied`	int	Antal rader som har kopierats.
`batch_count`	int	Antal batcher som bearbetats.
`elapsed_time`	float	Tiden som tas för operationen i sekunder.

Metodsignatur

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
)

Kolumnmappningar

Som standard mappar bulkcopy() kolumner efter ordinalposition. Varje datakolumn mappas till tabellkolumnen vid samma index. Använd parametern column_mappings för att åsidosätta detta beteende.

Kolumnnamnslista

Varje position i listan motsvarar källdataindexet:

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

Avancerat format: explicit indexkartläggning

Varje tuple har formen (source_index, target_column_name). Använd detta format för att hoppa över eller omordna kolumner:

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

Ladda från filer

Du kan ladda data från CSV-filer och andra filformat genom att skicka en generator till bulkcopy().

CSV-fil

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

Stora filer med batchning

Ställ in parametern batch_size för att styra hur många rader drivrutinen skickar per batch. Denna metod fungerar bra för stora filer:

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

Ladda pandas DataFrames

Konvertera en pandas DataFrame till en lista med tupler innan du skickar den till 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)

Hantera NULL-värden

Skicka None in valfri kolumnposition för att infoga ett SQL-värde NULL :

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)

Identitetskolumner

För att infoga explicita identitetsvärden, sätt 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)

När keep_identity=False (standardvärdet) utelämnar du identitetskolumnen i dina data och använder column_mappings för att ange de kolumner som inte är identitetskolumner.

Masskopieringsalternativ

Parameter	Standardinställning	Beskrivning
`batch_size`	`0`	Rader per sats. `0` låter servern välja den optimala storleken.
`timeout`	`30`	Tidsgräns för operationen i sekunder.
`keep_identity`	`False`	Bevara identitetsvärden från källdata.
`check_constraints`	`False`	Kontrollera tabellbegränsningar under belastningen.
`table_lock`	`False`	Skaffa ett bordsnivålås istället för radnivålås.
`keep_nulls`	`False`	Bevara NULL-värden istället för att infoga kolumnstandarder.
`fire_triggers`	`False`	Kör INSERT utlösare på måltabellen.
`use_internal_transaction`	`False`	Omslut varje batch med en intern transaktion.

Hantering av fel

bulkcopy() utlöser ett undantag om inläsningen misslyckas, så omge anropet med ett try/except-block för att fånga upp fel. Tänk på att bulkcopy() körs på sin egen interna anslutning och sparar de kopierade raderna oberoende, så en conn.rollback() på huvudanslutningen inte kan ångra dem. För att göra en batch atomisk anger du use_internal_transaction=True, som omger varje batch med en egen transaktion som automatiskt rullas tillbaka om batchen misslyckas:

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

För att grinda en last bakom din egen valideringslogik, bulkkopiera in i en staging-tabell och promovera sedan raderna till måltabellen med en INSERT ... SELECT inuti en transaktion på din huvudanslutning. Det INSERT körs via din anslutning, så conn.rollback() återställer det om valideringen misslyckas.

Authentication

Bulkkopiering använder en separat intern kanal som kräver sin egen token. Drivrutinen hanterar hämtning av token automatiskt för de autentiseringsmetoder som stöds.

Hanterad identitet (ActiveDirectoryMSI)

Användning Authentication=ActiveDirectoryMSI för systemtilldelad eller användartilldelad hanterad identitet. Denna autentiseringsmetod rekommenderas för Azure-hostade tjänster såsom Azure-VM:ar, App Service, Functions och 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")

För en användartilldelad hanterad identitet anger du klient-ID:t i anslutningssträngen:

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

Tjänstens huvudnamn (ActiveDirectoryServicePrincipal)

Använd Authentication=ActiveDirectoryServicePrincipal för autentisering med tjänstens huvudnamn (klientautentiseringsuppgifter).

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

Standardkedja för inloggningsuppgifter (ActiveDirectoryDefault)

ActiveDirectoryDefault försöker flera autentiseringsuppgiftsleverantörer i följd, till exempel miljövariabler, arbetsbelastningsidentitet, hanterad identitet med mera. Det fungerar både för lokal utveckling och Azure-hostade tjänster utan kodändringar.

För mer information om autentisering, se Microsoft Entra-autentisering.

Prestandatips

Följande tekniker hjälper dig att maximera masskopikapaciteten.

Använd generatorer för stora datamängder

Generatorer minimerar minnesanvändningen eftersom bulkcopy() de accepterar alla iterabla alternativ:

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

Använd bordslås för snabbare laster

När du inte har några samtidiga läsare, ställ table_lock=True in för att minska låsningsbelastning under stora initiala laster.

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

Inaktivera index under laddning

Inaktivera tillfälligt icke-grupperade index innan massinläsningen och bygg sedan om dem för förbättrad prestanda:

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

Ladda tabeller parallellt

Öppna en separat anslutning för varje tabell och kör lasterna samtidigt.

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

Jämförelse med alternativ

Följande tabell jämför bulkkopiering med andra metoder för datainsättning.

Metod	Användningsfall	Performance
`cursor.bulkcopy()`	Stora datamängder (mer än 1 000 rader).	Snabbaste
`cursor.executemany()`	Medelstora datamängder med parametrar.	Medel
`cursor.execute()` i en slinga	Små datamängder med enkel logik.	Långsammaste

Feedback

Var den här sidan till hjälp?

Last updated on 2026-07-22