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

Převeďte datový rámec pandas na seznam n-tic, než ho předáte do 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)

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í.
`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. Počínaje verzí mssql-python 1.12.0 dědí toto interní připojení časový limit připojení kurzoru: pokud jste funkci mssql_python.connect() předali hodnotu timeout=<seconds>, použije se při otevření připojení pro hromadné kopírování stejná kladná hodnota. Pokud jste ji nenastavili (nebo jste předali timeout=0), interní připojení použije výchozí 15sekundový časový limit pro připojení. Hodnota se zaznamená při volání bulkcopy(), takže pozdější změny v nadřazeném připojení neovlivní právě probíhající hromadné kopírování. Zvyšte časový limit připojení na mateřském připojení u pomalých, zpomalených nebo vysoce latencních koncových zařízení (například přes VPN nebo napříč 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í.

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()`	Velké datové sady (více než 1 000 řádků).	Nejrychlejší
`cursor.executemany()`	Střední datové sady s parametry.	Moderate
`cursor.execute()` v smyčce	Malé datové soubory s jednoduchou logikou.	Nejpomalejší

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-07-28