Használd az mssql-pythont az Apache Arrow-val

Az mssql-python driver Apache Arrow fetch módszereket biztosít a Microsoft SQL és Azure SQL Database nagy teljesítményű oszlopos adatkereséséhez.

Az Apache Arrow egy nyelvek közötti fejlesztő platform memórián belüli oszlopos adatokhoz. Az illezser közvetlenül az ODBC eredményhalmazokat alakítja át Arrow formátumba C++-ban, megkerülve a Python objektumalkotást a jobb teljesítmény érdekében.

Az Arrow integráció lehetővé teszi:

  • Másolásmentes adatátvitel a Polars, a pandas és a DuckDB számára. A „Zero-copy” azt jelenti, hogy az adatok egyetlen memóriapufferben maradnak, amelybe a meghajtóprogram ír, és amelyből a felhasználó könyvtárak közvetlenül olvasnak, így a sorok nem másolódnak át köztes Python-objektumokba.
  • Eredményhalmazok streamelése a(z) RecordBatchReader használatával, anélkül, hogy mindent betöltene a memóriába.
  • Oszlopos adatformátum, amely ideális analitikai és gépi tanulási feladatokhoz.
  • Csökkent memóriahasználat a soronkénti Python objektumalkotáshoz képest.

Kurzormetódusok

A pyarrow csomagnak Arrow fetch metódusok kell használnia. Telepítse a(z) pip install pyarrow használatával. Ha a(z) pyarrow nincs telepítve, bármely Arrow-metódus meghívása egy ImportError kivételt vált ki.

Az mssql-python illesztőprogram három metódust ad hozzá a kurzorobjektumhoz az Arrow adathozzáféréshez. Mindhárom módszer az ODBC eredményhalmazokat Arrow formátumra alakítja az illesztőprogram C++ rétegében, így elkerülve a köztes Python objektumok létrehozását.

  • arrow() az egész eredményhalmazt egy memórián belüli táblázatként adja vissza. A legegyszerűbben használható.
  • arrow_batch() egyszerre a sorok egy kötegét adja vissza, így manuálisan vezérelheted a ciklust.
  • arrow_reader() egy iterátort ad vissza, amely automatikusan adagokat ad. A legjobb nagy méretű eredmények továbbítására.

Az cursor.arrow(batch_size=8192) használata

Kérje le a teljes eredményhalmazt egyetlen pyarrow.Table-ként. Ez a módszer a legegyszerűbb, és jól működik, ha a teljes eredményhalmaz beilleszkedik a memóriába.

import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product")
table = cursor.arrow()

print(type(table))       # <class 'pyarrow.lib.Table'>
print(table.num_rows)    # Number of rows fetched
print(table.num_columns) # Number of columns
print(table.schema)      # Column names and Arrow types
print(table.to_pandas()) # Convert to pandas DataFrame

Megjegyzés:

Ha a kapcsolati karakterlánc Authentication=ActiveDirectoryDefault használ, az illesztőprogram DefaultAzureCredential használ, amely több hitelesítőadat-szolgáltatót próbál ki egymás után. Az első kapcsolat lassú lehet, mert az SDK végigjárja a láncot, amíg meg nem talál egy működő szolgáltatót. A termelésben, ha tudod, melyik hitelesítéstípust használja a környezeted, közvetlenül megadd (például ActiveDirectoryMSI menedzselt identitásnál), hogy elkerüld a láncos sétát. További információ: Microsoft Entra-hitelesítés.

Az cursor.arrow_batch(batch_size=8192) használata

Kérjen le egyetlen pyarrow.RecordBatch elemet, amely legfeljebb batch_size sort tartalmaz. Ezt a módszert olyan egyéni kötegelt feldolgozási ciklusokhoz használd, amelyeknél pontosan szeretnéd szabályozni, hogy egyszerre hány sort kérjen le.

cursor.execute("SELECT * FROM Production.TransactionHistory")

while True:
    batch = cursor.arrow_batch(batch_size=10000)
    if batch.num_rows == 0:
        break
    # Process each batch
    print(f"Fetched {batch.num_rows} rows")

Az cursor.arrow_reader(batch_size=8192) használata

Küldj vissza egy olvasót, amely RecordBatch objektumokat ad, amíg az eredményhalmaz ki nem merül. Ez a módszer a legnagyobb memória-hatékonyságú megoldás nagy eredményhalmazok esetén.

cursor.execute("SELECT * FROM Production.TransactionHistory")
reader = cursor.arrow_reader(batch_size=50000)

for batch in reader:
    # Process streaming batches without loading all data
    print(f"Batch: {batch.num_rows} rows")

Az olvasó az eredményeket a kapcsolaton keresztül streameli, így amíg egy olvasatlan olvasó nyitva van, az a kapcsolat nem indíthat el újabb állítást. Az egyik próbálkozás Connection is busy with results for another command hibával meghiúsul.

Három dolog szabadítja fel az olvasót: ha a végéig iterálunk rajta, ha bezárjuk a szülő kurzort, vagy ha bezárjuk az olvasót. Ha abbahagyod az olvasást, mielőtt az eredménykészlet kimerülne, és továbbra is használod a kurzort, zárd be az olvasót. A lezárás a szülőkurzort is visszaállítja, így egy másik utasítást futtathatsz rajta.

Használd az olvasót kontextuskezelőként, hogy akkor is záruljon, ha egy kivétel megszakítja a hurkot:

cursor.execute("SELECT * FROM Production.TransactionHistory")

rows_seen = 0
with cursor.arrow_reader(batch_size=50000) as reader:
    for batch in reader:
        rows_seen += batch.num_rows
        if rows_seen >= 100000:
            break

# The reader is closed here, and the cursor is ready for the next statement.
cursor.execute("SELECT COUNT(*) FROM Production.TransactionHistory")

Közvetlenül is hívhatsz reader.close() . Ha többször hívod meg, az is biztonságos, és a reader.closed tulajdonság jelzi, hogy bezártad-e.

Gyakori minták

Az Arrow táblázatok közvetlenül integrálódnak a népszerű Python adatkönyvtárakkal. Az alábbi példák bemutatják, hogyan lehet Arrow adatokat továbbítani panda-k, Polars-, DuckDB-nek és fájlformátumoknak anélkül, hogy adatokat másolnánk.

Eredmények betöltése a pandasba

cursor.execute("SELECT * FROM Production.Product")
table = cursor.arrow()

# Convert to pandas with zero-copy where possible
df = table.to_pandas()
print(df.head())

Eredmények betöltése Polarsba

import polars as pl

cursor.execute("SELECT * FROM Production.Product")
table = cursor.arrow()

df = pl.from_arrow(table)
print(df)

Lekérdezési eredmények a DuckDB-vel

A DuckDB közvetlenül SQL-ben képes lekérdezni az Arrow táblákat anélkül, hogy adatokat másolna. Ez a képesség hasznos, ha SQL-stílusú elemzésre van szükséged olyan eredményhalmazoknál, amelyek már Arrow formátumban vannak.

import duckdb

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
arrow_table = cursor.arrow()

# Query the Arrow table with DuckDB SQL
result = duckdb.sql("SELECT CustomerID, SUM(TotalDue) FROM arrow_table GROUP BY CustomerID")
print(result.fetchall())

Nagy eredményhalmazok továbbítása Parquet-be

Nagy eredményhalmazok esetén az Arrow-kötegeket közvetlenül egy Parquet-fájlba írhatja folyamatos adatátvitellel, anélkül hogy a teljes adathalmazt a memóriába kellene tölteni. A ParquetWriter az egyes kötegeket fokozatosan írja ki.

import pyarrow.parquet as pq

cursor.execute("SELECT * FROM Production.TransactionHistory")
reader = cursor.arrow_reader(batch_size=100000)

# Write streaming batches to a Parquet file
writer = None
for batch in reader:
    if writer is None:
        writer = pq.ParquetWriter("output.parquet", batch.schema)
    writer.write_batch(batch)

if writer:
    writer.close()

Exportálás más formátumokba

A PyArrow beépített írókat biztosít CSV-hez és az Arrow IPC fájlformátumhoz (más néven Feather V2). Az Arrow IPC fájlok pontosan megőrzik az Arrow típusokat, és gyorsan visszaolvashatók.

import pyarrow as pa
import pyarrow.csv as pcsv

cursor.execute("SELECT * FROM Production.Product")
table = cursor.arrow()

# Write to CSV
pcsv.write_csv(table, "products.csv")

# Write to an Arrow IPC file
with pa.ipc.new_file("products.arrow", table.schema) as writer:
    writer.write_table(table)

Töltsd be az Arrow adatait az SQL Server-be

A cursor.bulkcopy_arrow() metódus úgy ír Arrow-adatokat egy táblába, hogy azokat először nem konvertálja Python-sortuplákká. Az source érv elfogadja a következők bármelyikét:

  • Egy pyarrow.Table.

  • Egy pyarrow.RecordBatch.

  • Egy pyarrow.RecordBatchReader, beleértve a cursor.arrow_reader() által visszaadott beolvasót is.

  • Bármely objektum, amely az Arrow C adatinterfészt __arrow_c_array__ vagy __arrow_c_stream__ révén elérhetővé teszi.

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 ##SensorArchive (
        SensorID int NOT NULL,
        Reading float NULL,
        Location nvarchar(50) NULL
    )
""")

table = pa.table({
    "SensorID": pa.array([1, 2, 3], type=pa.int32()),
    "Reading": pa.array([20.5, None, 22.1], type=pa.float64()),
    "Location": pa.array(["Plant A", "Plant B", None], type=pa.string()),
})

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

Az Arrow null értékeket SQL NULL értékként írják.

Streamelj egy eredményhalmazt egy másik táblázatba

Mivel bulkcopy_arrow() elfogad egy olvasót, egy nagy eredményhalmazt lehet áthelyezni táblák között anélkül, hogy az memóriában megvalósulná:

cursor.execute("""
    CREATE TABLE ##ProductArchive (
        ProductID int NOT NULL,
        Name nvarchar(50) NOT NULL,
        ListPrice money NOT NULL
    )
""")

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product")

with cursor.arrow_reader(batch_size=100000) as reader:
    result = cursor.bulkcopy_arrow("##ProductArchive", reader, batch_size=100000)

print(f"Copied {result['rows_copied']} rows")

A Arrow típusok összehangolása a céloszlopokkal

Az Arrow író megköveteli, hogy minden Arrow oszloptípus kompatibilis legyen a célállomás SQL oszloptípusával. Nem konvertálódik családok között, így a páratlanság előfordul ValueError , mielőtt bármilyen sort írnának:

ValueError: Cannot map Arrow column 'ListPrice' (Float64) to SQL column 'ListPrice'
(Money): Usage Error: type combination is not supported by the Arrow row-major writer

Az Arrow-típus kiválasztásához használja fordított sorrendben a Data type mappings elemet. A pénznem, decimális és numerikus oszlopokhoz float64 szükséges, nem decimal128. A cursor.arrow() segítségével visszaolvasott adatok már eleve a megfelelő típusokkal rendelkeznek, így az SQL Serverből beolvasott tábla konverzió nélkül betölthető egy egyező táblába.

Térképoszlopok név szerint

Ha az Arrow oszlopainak sorrendje nem egyezik a céltábla oszlopsorrendjével, a column_mappings argumentumban adja meg a céltábla oszlopneveit az Arrow oszlopainak sorrendjében:

from decimal import Decimal

table = pa.table({
    "Name": pa.array(["Widget"], type=pa.string()),
    "ProductID": pa.array([9001], type=pa.int32()),
    "ListPrice": pa.array([Decimal("12.34")], type=pa.decimal128(19, 4)),
})

cursor.bulkcopy_arrow(
    "##ProductArchive",
    table,
    column_mappings=["Name", "ProductID", "ListPrice"],
)

A módszer ugyanazokat az opciókat fogadja, mint cursor.bulkcopy(), beleértve batch_size, timeout, keep_identity, table_lock, és keep_nulls. További információért ezekről a lehetőségekről lásd: Tömeges másolat.

Megjegyzés:

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

Adattípus-leképezések

Az Arrow fetch metódusok a Microsoft SQL típusokat a C++ szintű Arrow típusokhoz képezik.

Microsoft SQL-típus Nyíltípus
int, smallint, tinyint, bigint int32, int16, int8, int64
lebegő, valós float64, float32
decimális, numerikus decimal128
bit bool
char, varchar, nchar, nvarchar utf8
szöveg, ntext large_utf8
bináris, varbináris binary, large_binary
date date32
time time64[us]
datetime, datetime2, smalldatetime timestamp[us]
datetimeoffset timestamp[us, tz=UTC]
uniqueidentifier utf8 (nagybetűs karakterlánc)
xml utf8

Megjegyzés:

Az illesztőprogram átalakítja a datetimeoffset típust UTC-re, mert az Arrow oszlopokhoz fix időzónát kell írni. Az illesztőprogram az átalakítás során a Microsoft SQL-ből származó cellánkénti időzóna-információt UTC-re normalizálja.

Ezt sql_variant a típust nem támogatják az Arrow fetch metódusok, és nem támogatott adattípus kivételt generál. Használj standard fetchone(), fetchmany(), vagy fetchall() olyan lekérdezésekhez, amelyek oszlopokat adnak vissza sql_variant .

Teljesítménnyel kapcsolatos szempontok

Az Arrow fetch módszerek a leggyorsabbak analitikai és tömeges adatműveletekhez, míg a standard kurzoros módszerek jobban megfelelnek tranzakciós mintákra, ahol kis eredményhalmaz.

Mikor kell használni az Arrow-t a standard fetch ellen

Scenario Ajánlott megközelítés
Kérjen le néhány sort megjelenítéshez fetchone() / fetchall()
Adat töltése Pandákba vagy Polarokba cursor.arrow()
Nagy adathalmazok feldolgozása darabokban cursor.arrow_reader()
Egysoros lekérdezések vagy kis eredményhalmazok fetchone() / fetchval()
Elemző vagy aggregációs folyamatok cursor.arrow() + Polars/DuckDB
Eredmények írása Parquet vagy Arrow IPC formátumra cursor.arrow_reader() + PyArrow I/O

Memóriakezelés nagy adathalmazoknál

Azoknál az eredményhalmazoknál, amelyek meghaladhatják a rendelkezésre álló memóriát, használja a(z) arrow_reader() elemet ésszerű batch_size értékkel.

cursor.execute("SELECT * FROM Production.TransactionHistory")

# Process in batches of 100K rows
reader = cursor.arrow_reader(batch_size=100000)
total_rows = 0

for batch in reader:
    # Work with each batch individually
    total_rows += batch.num_rows
    # batch goes out of scope and memory is freed

print(f"Processed {total_rows} rows")

Hangolási tétel mérete

A batch_size paraméter szabályozza, hány sort hoznak le minden egyes tételben. Az optimális méret a sorszélességtől és a rendelkezésre álló memóriától függ. A nvarchar(max) vagy varbinary(max) típusú nagy oszlopokat tartalmazó szélesebb sorok számára a kisebb kötegméretek előnyösebbek, míg a keskeny sorok számára a nagyobbak.

  • Alapértelmezés (8192): Jó egyensúly a legtöbb munkaterheléshez.
  • Kisebb (1000-5000): Széles asztalokhoz és nagy oszlopokhoz használják.
  • Nagyobb (50000-100000): Szűk táblákhoz vagy akkor használat, amikor az áteresztés fontosabb, mint a memória.
# Narrow table with many rows - use larger batches
cursor.execute("SELECT ProductID, ListPrice FROM Production.Product")
table = cursor.arrow(batch_size=100000)

# Wide table with LOB columns - use smaller batches
cursor.execute("SELECT * FROM Production.Document")
table = cursor.arrow(batch_size=1000)

Kapcsolódó tartalom

  • Last updated on