Používejte hromadné kopírovaní s mssql-python

Ovladač mssql-python obsahuje funkci hromadného kopírování, která efektivně vkládá velké množství dat do SQL Server, Azure SQL Database, Azure SQL Managed Instance a SQL databáze v Microsoft Fabric.

Metoda cursor.bulkcopy() poskytuje vysoce výkonnou cestu pro načítání velkých datových sad:

• Minimalizuje to zpáteční cesty po síti.
• Volitelně obchází kontrolu omezení během zátěže.
• Používá optimalizovaný protokol TDS bulk insert.
• Dosáhne propustnosti srovnatelné s bcp.exe a SqlBulkCopy.

Nativní rozšíření založené mssql_py_core na Rustu pohání funkci hromadné kopírování. Běží mimo běžný kurzorový execute() kanál.

Základní použití

Na kurzoru zavolejte bulkcopy() a předejte název cílové tabulky a iterovatelný objekt řádkových n-tic nebo objektů Row:

Important

Pokud ve stejné relaci vytvoříte nebo upravíte cílovou tabulku, zavolejte conn.commit() před bulkcopy(). Protokol hromadné kopírování používá samostatný interní kanál pro čtení metadat tabulek, takže neregistrovaná změna DDL může způsobit zablokování nebo časový limit.

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

Návratová hodnota

bulkcopy() vrací slovník:

Key	Typ	Popis
rows_copied	int	Počet řádků úspěšně zkopírovaných.
batch_count	int	Počet zpracovaných šarží.
elapsed_time	float	Doba trvání operace v sekundách.

Metodní podpis

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
)

Mapování sloupců

Ve výchozím nastavení bulkcopy() jsou sloupce mapovány podle ordinální pozice. Každý datový sloupec odpovídá sloupci tabulky se stejným indexem. Pomocí parametru column_mappings potlačíte toto chování.

Seznam názvů sloupců

Každá pozice v seznamu odpovídá indexu zdrojových dat:

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

Pokročilý formát: explicitní mapování indexů

Každá n-tice má tvar (source_index, target_column_name). Použijte tento formát pro přeskočení nebo přeuspořádání sloupců:

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

Načítání ze souborů

Data z CSV souborů a dalších formátů můžete načíst tím, že generátor pošlete do bulkcopy().

Soubor CSV

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

Velké soubory s dávkovým zpracováním

Nastavte parametr batch_size tak, aby určoval, kolik řádků ovladač odešle v každé dávce. Tento přístup dobře funguje pro velké soubory:

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

Load pandas DataFrames

DataFrame má sloupcovou strukturu, takže nejrychlejší cesta je bulkcopy_arrow(), která využívá tabulku Arrow, kterou už pandas umí vytvořit. bulkcopy() přijímá řádkové n-tice, takže je nejdřív potřeba převést sloupce na objekty Pythonu.

Před načtením přeneste tabulku šipek na typy cílových sloupců. pyarrow odvodí float64 pro číselný sloupec, který ovladač nemůže namapovat na money, decimal nebo numeric:

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)

Bez castu zátěž selže s ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money). Vytvořte přetypování pomocí Table.cast() namísto předání schématu do Table.from_pandas(), protože Table.from_pandas() nedokáže přímo převést sloupec typu float na decimal128. NaN hodnoty se na této cestě stávají SQL NULL , takže je nemusíte nejdřív nahrazovat.

Pokud místo toho potřebujete cestu n-tice řádku, itertuples() už vrací n-tice, když předáte name=None:

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

Načíst data Apache Arrow

Použijte cursor.bulkcopy_arrow() k načtení dat z Apache Arrow. Tato metoda čte přímo z paměti Arrow, takže před jejím voláním nemusíte vytvářet řádkové n-tice Pythonu.

Argument source přijímá pyarrow.Table, pyarrow.RecordBatch, pyarrow.RecordBatchReader nebo jakýkoli objekt, který poskytuje datové rozhraní Arrow C. Zbývající argumenty jsou stejné jako .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")

Každý typ sloupce Arrow musí být kompatibilní s cílovým SQL sloupcem. Zapisovač nepřevádí mezi rodinami typů, takže předání sloupce float64 do sloupce money vyvolá ValueError ještě před zapsáním jakýchkoli řádků. Použijte decimal128 pro sloupce typu peníze, desetinné a číselné.

Předání zdroje Arrow do bulkcopy() vyvolá TypeError a přesměruje vás na bulkcopy_arrow().

Pro více informací o podpoře Arrow, včetně toho, jak streamovat výslednou sadu z jedné tabulky do druhé, viz integrace Apache Arrow.

Zpracovat hodnoty NULL

Předejte do libovolné pozice sloupce None pro vložení hodnoty SQL 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)

Identifikační sloupce

Pro vložení explicitních identitních hodnot nastavte 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)

Pokud je nastavena možnost keep_identity=False (ve výchozím nastavení), vynechejte z dat sloupec identit a pomocí column_mappings cílte na sloupce, které nejsou sloupci identit.

Možnosti hromadné kopie

Parameter	Výchozí	Popis
batch_size	0	Řady na várku. 0 umožňuje serveru vybrat optimální velikost.
timeout	30	Časový limit operace v sekundách. Platí to pro samotnou operaci hromadného kopírování, ne pro vnitřní spojení. Použijte 0 k vypnutí časového limitu operace.
keep_identity	False	Zachovat hodnoty identity ze zdrojových dat.
check_constraints	False	Kontrolovat omezení tabulky během načítání.
table_lock	False	Získejte zámek na úrovni tabulky místo zámků na úrovni řádků.
keep_nulls	False	Zachovejte hodnoty NULL místo vkládání výchozích sloupců.
fire_triggers	False	Spusťte INSERT triggery na cílové tabulce.
use_internal_transaction	False	Každou dávku zabalte do interní transakce.

Note

bulkcopy() otevírá samostatné interní připojení k serveru. Toto interní připojení dědí časový limit dotazu kurzoru: před vytvořením kurzoru nastavte Connection.timeout na kladnou hodnotu a stejná hodnota bude omezovat časový limit pokusu o připojení pro hromadné kopírování. Pokud je časový limit dotazu kurzoru 0, interní spojení použije svůj výchozí 15sekundový časový limit pro připojení. Kurzor přijímá hodnotu při jeho vytvoření, takže následná změna Connection.timeout neovlivní existující kurzor ani hromadnou kopii během letu. Zvyšte časový limit dotazu před vytvořením kurzoru pro pomalé, zpomalené nebo vysoce latenční koncové body (například přes VPN nebo mezi regiony).

Řešte chyby

bulkcopy() Vyvolá výjimku, pokud načítání selže, takže volání se zabalí do bloku try/except , aby se chyby zachytily. Mějte na paměti, že to bulkcopy() běží na vlastním interním připojení a commituje zkopírované řádky nezávisle, takže conn.rollback() na vašem hlavním spojení je nemůže zrušit. Aby byla dávka atomická, nastavte use_internal_transaction=True, která každou dávku uzavře do vlastní transakce, jež se automaticky zruší, pokud dávka selže:

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

Chcete-li podmínit načtení vlastní validační logikou, hromadně zkopírujte data do pracovní tabulky a potom přesuňte řádky do cílové tabulky pomocí INSERT ... SELECT v rámci transakce v hlavním připojení. To běží na vašem připojení, takže INSERT to zruší, pokud validace conn.rollback() selže.

Autentizace

Hromadné kopírování používá samostatný interní kanál, který vyžaduje vlastní token. Ovladač automaticky zajišťuje získávání tokenů pro podporované autentizační metody.

Spravovaná identita (ActiveDirectoryMSI)

Použití Authentication=ActiveDirectoryMSI pro systémově přiřazenou nebo uživatelem přiřazenou spravovanou identitu. Tato autentizační metoda je doporučena pro služby hostované v Azure, jako jsou Azure VM, App Service, Functions a 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")

Pro uživatelem přiřazenou spravovanou identitu předejte ID klienta do připojovací řetězec:

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

Service principal (ActiveDirectoryServicePrincipal)

Použití Authentication=ActiveDirectoryServicePrincipal pro autentizaci principálu služby (klientských přihlašovacích údajů).

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

Výchozí řetězec přihlašovacích údajů (ActiveDirectoryDefault)

ActiveDirectoryDefault zkouší více poskytovatelů přihlašovacích údajů v řadě, například proměnné prostředí, identitu pracovní zátěže, spravovanou identitu a další. Funguje jak pro lokální vývoj, tak pro služby hostované v Azure bez změn v kódu.

Pro více informací o autentizaci viz Microsoft Entra autentizace.

Tipy týkající se výkonu

Následující techniky vám pomohou maximalizovat objemovou propustnost kopií.

Začněte ze sloupcového zdroje

bulkcopy() přebírá iterovatelný objekt obsahující ntice řádků, takže každá hodnota musí existovat jako objekt Pythonu ještě před zahájením kopírování. Když už jsou data ve sloupcovém formátu, bulkcopy_arrow() přímo čte buffery Arrow a tento krok přeskočí. Datový rámec pandas nebo Polars, soubor Parquet a výsledek cursor.arrow() – to vše jsou zdroje Arrow. Pro více informací viz Načtení dat Apache Arrow.

Použijte generátory pro velké datové sady

Generátory minimalizují využití paměti, protože bulkcopy() přijímají jakékoli iterovatelné:

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

Uzamkněte tabulky pro rychlejší načítání

Pokud nemáte žádné souběžné čtenáře, nastavte table_lock=True, aby se během rozsáhlého počátečního načítání snížila režie spojená se zamykáním.

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

Vypnout indexy během načítání

Dočasně vypněte neklastrované indexy před hromadným načtením a poté je znovu sestavte pro lepší výkon:

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

Načítat tabulky paralelně

Otevřete samostatné připojení pro každou tabulku a spouštějte zátěže současně.

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

Srovnání s alternativami

Následující tabulka porovnává hromadnou kopii s jinými metodami vkládání dat.

Metoda	Případ použití	výkon
cursor.bulkcopy_arrow()	Velké datové soubory, které jsou už sloupcové.	Nejrychlejší
cursor.bulkcopy()	Velké datové sady (více než 1 000 řádků) z řádkově orientovaných zdrojů.	Rychlé
cursor.executemany()	Střední datové sady s parametry.	Moderate
cursor.execute() v smyčce	Malé datové soubory s jednoduchou logikou.	Nejpomalejší

Související obsah

• Spouštění dotazů pomocí mssql-python
• Mapování datových typů pro mssql-python
• Správa transakcí pomocí mssql-python
• Ladění výkonu aplikací mssql-python