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

En DataFrame är kolumnformad, så den snabbaste vägen är bulkcopy_arrow(), vilket förbrukar Arrow-tabellen som pandas redan vet hur man producerar. bulkcopy() accepterar radtupler, så du måste först omvandla kolumnerna till Python-objekt.

Konvertera Arrow-tabellen till målkolumnernas datatyper innan du läser in den. pyarrow härleder float64 för en numerisk kolumn, som föraren inte kan mappa till pengar, decimal eller numerisk:

import pandas as pd
import pyarrow as pa
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()

target = pa.schema([
    pa.field('ID', pa.int32()),
    pa.field('Name', pa.string()),
    pa.field('Amount', pa.decimal128(19, 4)),   # MONEY
])

table = pa.Table.from_pandas(df, preserve_index=False).cast(target)
result = cursor.bulkcopy_arrow("##PandasDemo", table)

Utan kastet misslyckas lasten med ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money). Skapa typkonverteringen med Table.cast() i stället för att skicka schemat till Table.from_pandas(), som inte kan konvertera en kolumn med flyttal direkt till decimal128. NaN värden blir SQL NULL på denna väg, så du behöver inte ersätta dem först.

Om du behöver sökvägen för radtupler i stället, returnerar itertuples() redan tupler när du skickar in name=None:

data = list(df.itertuples(index=False, name=None))
result = cursor.bulkcopy("##PandasDemo", data)

Ladda data från Apache Arrow

Använd cursor.bulkcopy_arrow() för att ladda Apache Arrow-data. Denna metod läser direkt från Arrow-minnet, så du bygger inte Python-radtupler innan du anropar den.

Argumentet source accepterar en pyarrow.Table, en pyarrow.RecordBatch, en pyarrow.RecordBatchReader eller ett valfritt objekt som exponerar Arrow C-datagränssnittet. De återstående argumenten är desamma som bulkcopy().

import mssql_python
import pyarrow as pa

conn = mssql_python.connect(connection_string)

# bulkcopy_arrow() opens its own connection, so commit the table creation first.
conn.autocommit = True
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##ArrowDemo (ID INT, Name NVARCHAR(50), Amount FLOAT)
""")

table = pa.table({
    "ID": pa.array([1, 2, 3], type=pa.int32()),
    "Name": pa.array(["Alice", "Bob", "Carol"], type=pa.string()),
    "Amount": pa.array([50000.0, 60000.0, 55000.0], type=pa.float64()),
})

result = cursor.bulkcopy_arrow("##ArrowDemo", table)
print(f"Copied {result['rows_copied']} rows")

Varje Arrow-kolumntyp måste vara kompatibel med sin destinations-SQL-kolumntyp. Författaren konverterar inte mellan typsnittsfamiljer, så att skicka en float64 kolumn till en pengakolumn höjer ValueError innan några rader skrivs. Använd decimal128 för pengar, decimala och numeriska kolumner.

Att skicka en Arrow-källa till bulkcopy() utlöser TypeError och dirigerar dig till bulkcopy_arrow().

För mer information om Arrow-stöd, inklusive hur man strömmar en resultatmängd från en tabell till en annan, se Apache Arrow-integration.

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. Gäller för själva bulkkopieringsoperationen, inte för den interna anslutningen. Använd 0 för att inaktivera drifttimeouten.
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.

Note

bulkcopy() öppnar en separat intern anslutning till servern. Den interna anslutningen ärver markörens timeoutvärde för frågan: ställ in Connection.timeout på ett positivt värde innan du skapar markören, så begränsar samma värde även anslutningsförsöket för bulkkopiering. Om markörens frågetimeout är 0, använder den interna anslutningen sin standard 15-sekunders anslutningstimeout. En markör tar värdet när den skapas, så att ändra Connection.timeout i efterhand påverkar inte en befintlig markör eller en bulkkopia under flygning. Höj tidsgränsen för frågan innan du skapar markören för långsamma, strypta eller höglatenta slutpunkter (till exempel via VPN eller mellan olika regioner).

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.

Börja från en kolumnkälla

bulkcopy()tar en iterabel av radtupler, så varje värde måste existera som ett Python-objekt innan kopieringen startar. När datan redan är kolumnär bulkcopy_arrow() läser pilbuffrarna direkt och hoppar över det steget. En pandas- eller Polars-dataframe, en Parquet-fil och resultatet av cursor.arrow() är alla Arrow-datakällor. För mer information, se Ladda Apache Arrow-data.

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_arrow()	Stora datamängder som redan är kolumnära.	Snabbaste
cursor.bulkcopy()	Stora datamängder (mer än 1 000 rader) från radorienterade källor.	Snabbt
cursor.executemany()	Medelstora datamängder med parametrar.	Medel
cursor.execute() i en slinga	Små datamängder med enkel logik.	Långsammaste

Relaterat innehåll

• Exekver frågor med mssql-python
• Datatypmappningar för mssql-python
• Transaktionshantering med mssql-python
• Prestandaoptimering för mssql-python-applikationer