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

A DataFrame oszlopos felépítésű, így a leggyorsabb megoldás a bulkcopy_arrow(), amely közvetlenül felhasználja azt az Arrow-táblát, amelyet a pandas eleve elő tud állítani. bulkcopy() sortuplákat fogad el, ezért először Python-objektumokká kell kilapítani az oszlopokat.

Az Arrow-táblát alakítsd a céloszlopok adattípusaira, mielőtt betöltöd. pyarrow Egy float64 numerikus oszlopra következtet, amelyet a meghajtó nem tud pénzre, tizedesre vagy számra leképezni:

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)

A típusátalakítás nélkül a betöltés a(z) ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money) elemmel sikertelen. A castot inkább Table.cast() hozd létre, ahelyett hogy a sémát a Table.from_pandas() részére adnád át, mivel az nem tud közvetlenül egy float típusú oszlopot decimal128 típussá alakítani. NaN Az értékek ezen az úton SQL-vé NULL válnak, így nem kell először helyettesítened őket.

Ha inkább a sor-tuple útvonalra van szükséged, a(z) itertuples() már tuple-öket ad vissza, ha átadod a(z) name=None argumentumot:

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

Apache Arrow adatai betöltése

A(z) cursor.bulkcopy_arrow() használatával Apache Arrow-adatok tölthetők be. Ez a módszer közvetlenül az Arrow memóriájából olvas, így a meghívása előtt nem kell Python-sortuplákat létrehozni.

A(z) source argumentum elfogad egy pyarrow.Table, egy pyarrow.RecordBatch, egy pyarrow.RecordBatchReader vagy bármely olyan objektumot, amely biztosítja az Arrow C adatinterfészt. A fennmaradó érvek ugyanazok, mint 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")

Minden Arrow oszloptípusnak kompatibilisnek kell lennie a célállomás SQL oszloptípusával. Az író nem konvertál adattípus-családok között, így ha egy float64 oszlopot egy money oszlopnak adunk át, az még bármely sor kiírása előtt ValueError hibát eredményez. Pénz, tizedes és számos oszlopok használatadecimal128.

Ha egy Arrow-forrást ad meg, a bulkcopy()TypeError hibát vált ki, és a bulkcopy_arrow() oldalra irányítja.

További információkért az Arrow támogatásáról, beleértve az eredményhalmazi streamelést egyik táblából a másikba, lásd az Apache Arrow integrációját.

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. Magára a tömeges másolási műveletre vonatkozik, nem a belső kapcsolatra. A műveleti időtúllépés letiltásához használja a(z) 0 elemet.
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.

Note

bulkcopy() külön belső kapcsolatot nyit a szerverhez. Ez a belső kapcsolat örökli a kurzor lekérdezési időkorlátját: állítsa a(z) Connection.timeout értékét pozitívra a kurzor létrehozása előtt, és ugyanez az érték korlátozza a tömeges másoláshoz használt kapcsolat létrehozási kísérletét. Ha a kurzor lekérdezési időkorlátja 0, a belső kapcsolat az alapértelmezett, 15 másodperces kapcsolódási időkorlátját használja. A kurzor a létrehozásakor kapja meg az értéket, ezért a Connection.timeout későbbi módosítása nincs hatással a meglévő kurzorra vagy a folyamatban lévő tömeges másolási műveletre. Emeld meg a lekérdezési időtúlzást, mielőtt létrehoznád a kurzort lassú, korlátozott vagy nagy késleltetésű végpontokhoz (például VPN-en keresztül vagy régiók között).

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.

Kezdjük oszlopos forrásból

bulkcopy() egy sorokat reprezentáló tuple-ökből álló iterálhatót vár, ezért minden értéknek Python-objektumként kell léteznie, mielőtt a másolás megkezdődik. Amikor az adatok már oszloposak, bulkcopy_arrow() közvetlenül olvassa az Arrow puffereket, és kihagyja ezt a lépést. Egy pandas- vagy Polars-DataFrame, egy Parquet-fájl és a cursor.arrow() eredménye mind Arrow-forrás. További információért lásd: Apache Arrow adatai töltése.

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_arrow()	Nagy adathalmazok, amelyek már oszloposak.	Leggyorsabb
cursor.bulkcopy()	Nagy adathalmazok (több mint 1000 sor) sororientált forrásokból.	Gyors
cursor.executemany()	Közepes méretű adathalmazok paraméterekkel.	Mérsékelt
cursor.execute() ciklusban	Kis adathalmazok egyszerű logikával.	Leglassabb

Kapcsolódó tartalom

• Execute queries with mssql-python
• Adattípus-leképezések mssql-python esetén
• Tranzakciókezelés mssql-python segítségével
• Teljesítmény hangolás mssql-python alkalmazásokhoz