Gebruik mssql-python met Apache Arrow

De mssql-python-driver biedt Apache Arrow-ophaalmethoden voor hoogpresterende kolomgegevensopvraging uit Microsoft SQL en Azure SQL Database.

Apache Arrow is een platform voor taaloverstijgende ontwikkeling voor kolomgegevens in het geheugen. De driver zet ODBC-resultaatsets direct om in het Arrow-formaat in C++, waarbij Python-objectcreatie wordt omzeild voor betere prestaties.

Arrow-integratie maakt het volgende mogelijk:

• Zero-copy gegevensoverdracht naar Polars, pandas en DuckDB. "Zero-copy" betekent dat de data in één enkele geheugenbuffer blijft die de driver schrijft en die de consumerende bibliotheken direct lezen, zodat er geen rijen worden gedupliceerd in intermediaire Python-objecten.
• Resultaatsets streamen via RecordBatchReader zonder alles in het geheugen te laden.
• Columnar dataformaat ideaal voor analytics- en machine learning-workloads.
• Minder geheugengebruik vergeleken met het opmaken van Python-objecten per rij.

Cursormethoden

Het pyarrow pakket moet Arrow-ophaalmethoden gebruiken. Installeer het met pip install pyarrow. Als pyarrow niet geïnstalleerd is, genereert het aanroepen van een Arrow-methode een ImportError.

De mssql-python-driver voegt drie methoden toe aan het cursorobject voor Arrow-data-toegang. Alle drie de methoden converteren ODBC-resultatensets naar Arrow-formaat in de C++-laag van de driver, waardoor het creëren van tussenliggende Python-objecten voorkomt.

• arrow() geeft de volledige resultaatset terug als één in-memory tabel. Zijn het eenvoudigst te gebruiken.
• arrow_batch() geeft één batch rijen tegelijk terug, waardoor je handmatige controle over de lus hebt.
• arrow_reader() geeft een iterator terug die automatisch batches oplevert. Het beste voor het streamen van grote resultaten.

Het gebruiken van cursor.arrow(batch_size=8192)

Haal de volledige resultaatset op als één enkele pyarrow.Table. Deze methode is de eenvoudigste en werkt goed wanneer de volledige resultaatset in het geheugen past.

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

Het gebruiken van cursor.arrow_batch(batch_size=8192)

Haal één pyarrow.RecordBatch op met maximaal batch_size rijen. Gebruik deze methode voor aangepaste batchverwerkingslussen waarbij je fijnmazige controle nodig hebt over hoeveel rijen tegelijk worden opgehaald.

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

Het gebruiken van cursor.arrow_reader(batch_size=8192)

Geef een lezer terug die objecten oplevert RecordBatch totdat de resultaatset is uitgeput. Deze methode is de meest geheugen-efficiënte optie voor grote resultaatsets.

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

De lezer streamt resultaten via de verbinding, dus zolang een ongelezen lezer open is, kan die verbinding geen nieuwe uitspraak starten. Een poging mislukt met een Connection is busy with results for another command foutmelding.

Drie dingen laten de lezer los: het herhalen tot het einde, het sluiten van de oudercursor, of het sluiten van de lezer. Als je stopt met lezen voordat de resultaatset is uitgeput en de cursor blijft gebruiken, sluit dan de lezer. Het sluiten reset ook de oudercursor, zodat je er een andere instructie op kunt uitvoeren.

Gebruik de lezer als contextmanager zodat deze sluit, zelfs als een uitzondering de lus onderbreekt:

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

Je kunt ook reader.close() direct bellen. Het is veilig om het meer dan één keer aan te roepen, en de eigenschap reader.closed geeft aan of je het hebt gesloten.

Algemene patronen

Arrow-tabellen integreren direct met populaire Python-databibliotheken. De volgende voorbeelden laten zien hoe je Arrow-gegevens kunt doorgeven aan pandas, Polars, DuckDB en bestandsformaten zonder data te kopiëren.

Laad resultaten in pandas

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

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

Resultaten laden in Polars

import polars as pl

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

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

Zoekresultaten met DuckDB

DuckDB kan Arrow-tabellen direct in SQL bevragen zonder data te kopiëren. Deze mogelijkheid is handig wanneer je SQL-achtige analyses nodig hebt op resultaatsets die al in Arrow-formaat zijn.

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

Grote resultaatsets naar Parquet streamen

Voor grote resultaatsets streamt u Arrow-batches direct naar een Parquet-bestand zonder de volledige dataset in het geheugen te laden. De ParquetWriter schrijft elke batch incrementeel.

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 naar andere formaten

PyArrow biedt ingebouwde schrijvers voor CSV en het Arrow IPC-bestandsformaat (ook bekend als Feather V2). Arrow IPC-bestanden behouden Arrow-types exact en zijn snel terug te lezen.

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)

Laad Arrow-gegevens in SQL Server

De cursor.bulkcopy_arrow() methode schrijft Arrow-gegevens naar een tabel zonder deze eerst om te zetten in Python-rij-tuples. Het source argument accepteert een van de volgende elementen:

• A pyarrow.Table.

• A pyarrow.RecordBatch.

• Een pyarrow.RecordBatchReader, inclusief de lezer die wordt geretourneerd door cursor.arrow_reader().

• Elk object dat de Arrow C-data-interface blootstelt via __arrow_c_stream__ of __arrow_c_array__.

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

Pijl-nullwaarden worden geschreven als SQL NULL-waarden.

Stream een resultaatset naar een andere tabel

Omdat bulkcopy_arrow() een lezer accepteert, kun je een grote resultaatset tussen tabellen verplaatsen zonder dat het in het geheugen wordt gematerialiseerd:

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

Koppel Arrow-typen aan doelkolommen

De Arrow-schrijver vereist dat elk Arrow-kolomtype compatibel is met het bestemmings-SQL-kolomtype. Het converteert niet tussen families, dus er ontstaat ValueError een mismatch voordat er rijen zijn geschreven:

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

Gebruik de toewijzingen in Toewijzingen van gegevenstypen in omgekeerde volgorde om het Arrow-type te kiezen. geld, decimale en numerieke kolommen hebben float64, niet decimal128. Gegevens die met cursor.arrow() worden teruggelezen, hebben al de juiste typen, dus een tabel die uit SQL Server wordt gelezen, kan zonder conversie in een overeenkomende tabel worden geladen.

Kaartkolommen op naam

Wanneer de kolomvolgorde van de pijl niet overeenkomt met de bestemmingstabel, geef column_mappings dan de bestemmingskolomnamen door in de volgorde van pijlkolommen:

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

De methode accepteert dezelfde opties als cursor.bulkcopy(), waaronder batch_size, timeout, , keep_identity, table_lock, en keep_nulls. Voor meer informatie over deze opties, zie Bulk copy.

Gegevenstypetoewijzingen

De Arrow-ophaalmethoden mappen Microsoft SQL-types toe aan Arrow-types op C++-niveau.

Prestatie-overwegingen

Arrow fetch-methoden zijn het snelst voor analytics en bulkdata-operaties, terwijl standaard cursormethoden beter geschikt zijn voor transactionele patronen met kleine resultatensets.

Wanneer te gebruiken met Arrow versus standaard fetch

Geheugenbeheer voor grote datasets

Voor resultaatsets die mogelijk het beschikbare geheugen overschrijden, gebruik arrow_reader() met een redelijke batch_size.

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

Batchgrootte afstemmen

De batch_size parameter bepaalt hoeveel rijen er per batch worden opgehaald. De optimale grootte hangt af van je rijbreedte en beschikbare geheugen. Bredere rijen met grote kolommen zoals nvarchar(max) of varbinary(max) profiteren van kleinere batchgroottes, terwijl smalle rijen profiteren van grotere.

• Standaard (8192): Goede balans voor de meeste werklasten.
• Kleiner (1000-5000): Gebruik voor brede tabellen met grote kolommen.
• Groter (50000-100000): Gebruik voor smalle tabellen of wanneer doorvoer belangrijker is dan geheugen.

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

Verwante onderwerpen

• Gebruik mssql-python met Polars
• Gebruik mssql-python met pandas
• Gegevens ophalen met mssql-python
• Datatype-mappings voor mssql-python
• Prestatie-tuning voor mssql-python-applicaties