Használj tömeges másolatot mssql-python segítségével

Az mssql-python illesztőprogram tartalmaz egy tömeges másolási funkciót, amely hatékonyan helyez el nagy mennyiségű adatot az SQL Server-be, Azure SQL Database-be, Azure SQL Managed Instance-ba és SQL adatbázisba a Microsoft Fabric-ben.

A cursor.bulkcopy() módszer nagy teljesítményű útvonalat biztosít a nagy adathalmazok betöltésére:

• Minimalizálja a hálózati átmeneteket.
• Opcionálisan megkerüli a korlátozásellenőrzést terhelés alatt.
• Az optimalizált TDS tömeges beszúrási protokollt használja.
• Az áteresztőképessége összemérhető a(z) bcp.exe és SqlBulkCopy értékével.

A Rust-alapú mssql_py_core natív kiterjesztés működteti a tömeges másolat funkciót. A normál kurzorvezetéken execute() kívül fut.

Alapszintű használat

Hívd meg a bulkcopy() függvényt egy kurzoron, átadva a céltábla nevét, valamint egy sor-tuple-öket vagy Row objektumokat tartalmazó iterálható objektumot:

Important

Ha ugyanabban a munkamenetben hozod létre vagy módosítod a céltáblát, hívd meg a conn.commit() elemet a bulkcopy() előtt. A tömeges másolási protokoll egy külön belső csatornát használ a tábla metaadatok olvasásához, így egy elkötelezetlen DDL-változtatás holtpontot vagy időkorlátot okozhat.

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

Visszaadott érték

bulkcopy() szótárat ad vissza:

Key	Típus	Leírás
rows_copied	int	A sikeresen másolt sorok száma.
batch_count	int	A feldolgozott adagok száma.
elapsed_time	float	Az operáció ideje másodpercekben telt.

Metódusszignatúra

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
)

Oszlopleképezések

Alapértelmezés szerint az bulkcopy() oszlopokat sorrendhelyzet szerint térképezik le. Minden adatoszlop ugyanabban az indexben lévő táblázatoszlophoz rendelődik. Használd a column_mappings paramétert a viselkedés felülbírálására.

Oszlopnevek listája

A lista minden pozíciója megfelel a forrásadat indexnek:

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

Fejlett formátum: egyértelmű indexleképezés

Minden tuple a következő alakú: (source_index, target_column_name). Ezt a formátumot használd az oszlopok átugrására vagy átrendezésére:

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

Betöltés fájlokból

CSV fájlokból és más fájlformátumokból tölthetsz adatokat, ha egy generátort továbbítasz a .bulkcopy()

CSV-fájl

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

Nagy fájlok kötegeséssel

Állítsd be a batch_size paramétert úgy, hogy szabályozza, hány sort küld a meghajtó egy adagonként. Ez a megközelítés nagy fájlok esetén jól működik:

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

Pandas DataFrame-ek betöltése

Alakítsa át a pandas DataFrame-et tuple-ök listájává, mielőtt átadja a bulkcopy() számára:

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)

Kezeld a NULL értékeket

Passzolj None be bármelyik oszloppozíciót, hogy SQL NULL értéket adj be:

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)

Azonosító oszlopok

Az explicit módon megadott azonosítóértékek beszúrásához állítsa be a(z) 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)

Amikor keep_identity=False (az alapértelmezett), hagyd ki az identitásoszlopot az adataidból, és használd column_mappings a nem identitásoszlopok célzására.

Tömeges másolási beállítások

Paraméter	Default	Leírás
batch_size	0	Sorok kötegenként. 0 Lehetővé teszi, hogy a szerver válassza ki az optimális méretet.
timeout	30	A művelet időkorlátja másodpercben.
keep_identity	False	Őrizze meg az identitásértékeket a forrásadatból.
check_constraints	False	Ellenőrizd a táblázatkorlátokat a terhelés alatt.
table_lock	False	Szerezz egy táblázatszintű zárat a sorszintű zárak helyett.
keep_nulls	False	Őrizze meg a NULL értékeket az oszlop alapértelmezett értékeinek beszúrása helyett.
fire_triggers	False	Aktiválja a(z) INSERT triggert a céltáblán.
use_internal_transaction	False	Minden kötetet belső tranzakcióba csomagolj.

Hibák kezelése

bulkcopy() kivételt dob, ha a betöltés sikertelen, ezért a hibák elkapásához foglalja a hívást egy try/except blokkba. Ne feledd, hogy a bulkcopy() saját belső kapcsolaton fut, és önállóan véglegesíti a másolt sorokat, így a fő kapcsolaton végrehajtott conn.rollback() nem tudja visszavonni őket. Ahhoz, hogy egy köteg atomi legyen, állítsa be a(z) use_internal_transaction=True értéket, amely minden köteget a saját tranzakciójába foglal, és amely automatikusan visszagörgetődik, ha a köteg sikertelen lesz:

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

Ha egy betöltést a saját ellenőrzési logikádhoz akarsz kötni, tömegesen másold be az adatokat egy átmeneti táblába, majd a sorokat egy INSERT ... SELECT segítségével helyezd át a céltáblába, a fő adatbázis-kapcsolatodon belüli tranzakción belül. Ez INSERT a kapcsolatodon fut, ezért conn.rollback() visszavonja azt, ha az ellenőrzés sikertelen.

Authentication

A tömeges másolat külön belső csatornát használ, amelyhez saját tokent kell használni. Az illesztőprogram automatikusan kezeli a token beszerzést a támogatott hitelesítési módszerek esetében.

Kezelt identitás (ActiveDirectoryMSI)

Használja a(z) Authentication=ActiveDirectoryMSI elemet rendszer által hozzárendelt vagy felhasználó által hozzárendelt menedzselt identitáshoz. Ezt a hitelesítési módszert Azure-ban üzemeltetett szolgáltatásokhoz ajánlják, mint például az Azure VM-ek, App Service, Functions és 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")

Egy felhasználó által hozzárendelt menedzselt identitás esetén adjuk át az ügyfél azonosítót a kapcsolati karakterlánc-ben:

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

Szolgáltatásnév (ActiveDirectoryServicePrincipal)

Használja a(z) Authentication=ActiveDirectoryServicePrincipal elemet szolgáltatásfőnévvel (kliens hitelesítő adataival végzett) hitelesítéshez.

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

Hitelesítési adatok alapértelmezett lánca (ActiveDirectoryDefault)

ActiveDirectoryDefault Több hitelesítésszolgáltatót próbál egymás után, például környezeti változókat, munkaterhelési identitást, kezelt identitást és még sok mást. Helyi fejlesztéshez és Azure-ban üzemeltetett szolgáltatásokhoz is működik, kódváltoztatás nélkül.

További információért a hitelesítésről lásd: Microsoft Entra hitelesítés.

Teljesítménnyel kapcsolatos tippek

A következő technikák segítenek maximalizálni a tömeges szövegátvitelt.

Generátorok használata nagy adathalmazokhoz

A generátorok minimalizálják a memóriahasználatot, mert bulkcopy() elfogadnak bármilyen iterablet:

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

Gyorsabb terhelésekhez használj asztalzárokat

Ha nincsenek egyidejű olvasók, állítsd table_lock=True be úgy, hogy csökkentsd a zárolási túlterhelést nagy kezdeti terhelések alatt.

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

Indexek letiltása betöltés közben

Ideiglenesen tiltsd le a nem klaszterezett indexeket a tömeges betöltés előtt, majd utána építsd újra a jobb teljesítmény érdekében:

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

Táblák betöltése párhuzamosan

Nyiss külön kapcsolatot minden táblához, és egyszerre futtasd a terheléseket.

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

Összehasonlítás alternatívákkal

Az alábbi táblázat összehasonlítja a tömeges másolatot más adatbeillesztési módszerekkel.

Módszer	Felhasználási eset	Teljesítmény
cursor.bulkcopy()	Nagy adathalmazok (több mint 1000 sor).	Leggyorsabb
cursor.executemany()	Közepes méretű adathalmazok paraméterekkel.	Mérsékelt
cursor.execute() ciklusban	Kis adathalmazok egyszerű logikával.	Leglassabb

Kapcsolódó tartalom

• Lekérdezések végrehajtása
• Adattípus-leképezések
• Tranzakciókezelés
• Teljesítmény finomhangolása