Använd mssql-python med Apache Arrow

mssql-python-drivrutinen tillhandahåller Apache Arrow-hämtningsmetoder för högpresterande kolumnardata från Microsoft SQL och Azure SQL Database.

Apache Arrow är en plattform för utveckling av flera språk för kolumndata i minnet. Drivrutinen konverterar ODBC-resultatuppsättningar direkt till Arrow-format i C++, vilket undviker Python-objektskapande för bättre prestanda.

Integrering med Arrow möjliggör:

  • Nollkopieöverföring av data till Polars, Pandas och DuckDB. "Zero-copy" innebär att datan stannar i en enda minnesbuffert som drivrutinen skriver och konsumtionsbibliotek läser direkt, så inga rader dupliceras till mellanliggande Python-objekt.
  • Strömma resultatmängder genom RecordBatchReader utan att ladda allt i minnet.
  • Kolonnformat är idealiskt för analys- och maskininlärningsarbetsbelastningar.
  • Minskad minnesanvändning jämfört med rad-för-rad Python-objektskapande.

Markörmetoder

Paketet pyarrow måste använda Arrow-hämtningsmetoder. Installera den med pip install pyarrow. Om pyarrow inte är installerat resulterar anrop till valfri Arrow-metod i ett ImportError.

mssql-python-drivrutinen lägger till tre metoder till markörobjektet för åtkomst till Arrow-data. Alla tre metoder konverterar ODBC-resultatuppsättningar till Arrow-format i drivrutinens C++-lager, vilket undviker att skapa mellanliggande Python-objekt.

  • arrow() returnerar hela resultatuppsättningen som en minnestabell. Enklast att använda.
  • arrow_batch() returnerar en batch rader åt gången, vilket ger dig manuell kontroll över loopen.
  • arrow_reader() returnerar en iterator som automatiskt ger batcher. Bäst för att streama stora resultat.

Använda cursor.arrow(batch_size=8192)

Hämta hela resultatmängden som en enda pyarrow.Table. Denna metod är enklast och fungerar bra när hela resultatuppsättningen får plats i minnet.

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

Note

Om din anslutningssträng använder Authentication=ActiveDirectoryDefault, använder drivrutinen DefaultAzureCredential, som försöker med flera autentiseringsleverantörer i följd. Den första anslutningen kan vara långsam eftersom SDK:n går igenom kedjan tills den hittar en fungerande leverantör. I produktion, om du vet vilken typ av behörighet din miljö använder, ange det direkt (till exempel ActiveDirectoryMSI för managed identity) för att undvika kedjevandring. Mer information finns i Microsoft Entra-autentisering.

Använda cursor.arrow_batch(batch_size=8192)

Hämta en enda pyarrow.RecordBatch med upp till batch_size rader. Använd denna metod för anpassade batchbearbetningsloopar där du behöver finjusterad kontroll över hur många rader som hämtas åt gången.

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

Använda cursor.arrow_reader(batch_size=8192)

Returnera en läsare som returnerar RecordBatch-objekt tills resultatuppsättningen är uttömd. Denna metod är det mest minneseffektiva alternativet för stora resultatmängder.

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

Läsaren strömmar resultat över anslutningen, så medan en oläst läsare är öppen kan den anslutningen inte starta ett nytt uttalande. Ett försök misslyckas med ett Connection is busy with results for another command fel.

Tre saker frikopplar läsaren: att iterera igenom den till slutet, stänga den överordnade markören eller stänga läsaren. Om du slutar läsa innan resultatuppsättningen är slut och fortsätter använda markören, stäng läsaren. När du stänger den återställs också den överordnade markören, så att du kan köra en annan fråga med den.

Använd läsaren som kontexthanterare så att den stängs även om ett undantag avbryter loopen:

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

Du kan också ringa reader.close() direkt. Det är säkert att anropa den mer än en gång, och reader.closed-egenskapen anger om du har stängt den.

Vanliga mönster

Arrow-tabeller integreras direkt med populära Python-databibliotek. Följande exempel visar hur man skickar Arrow-data till pandas, Polars, DuckDB och filformat utan att kopiera data.

Ladda resultaten i 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())

Läs in resultat i Polars

import polars as pl

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

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

Sökresultat med DuckDB

DuckDB kan fråga Arrow-tabeller direkt i SQL utan att kopiera data. Denna funktion är användbar när du behöver SQL-liknande analys på resultatuppsättningar som redan är i Arrow-format.

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

Strömma stora resultatuppsättningar till Parquet

För stora resultatmängder, strömma Arrow-batcher direkt till en Parquet-fil utan att ladda hela datamängden i minnet. ParquetWriter skriver varje sats inkrementellt.

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 till andra format

PyArrow tillhandahåller inbyggda skrivare för CSV och filformatet Arrow IPC (även känt som Feather V2). Arrow IPC-filer bevarar Arrow-typer exakt och är snabba att läsa tillbaka.

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)

Ladda Arrow-data i SQL Server

Metoden cursor.bulkcopy_arrow() skriver Arrow-data till en tabell utan att först konvertera den till Python-radtupler. Argumentet source accepterar något av följande:

  • En pyarrow.Table.

  • En pyarrow.RecordBatch.

  • En pyarrow.RecordBatchReader, inklusive läsaren som returneras av cursor.arrow_reader().

  • Alla objekt som exponerar Arrow C:s datagränssnitt via __arrow_c_stream__ eller __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")

Pil-nullvärden skrivs som SQL NULL-värden.

Strömma en resultatuppsättning till en annan tabell

Eftersom bulkcopy_arrow() accepterar en läsare kan du flytta en stor resultatmängd mellan tabeller utan att materialisera den i minnet:

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

Matcha piltyper med destinationskolumnerna

Arrow-skrivaren kräver att varje Arrow-kolumntyp är kompatibel med sin destinations-SQL-kolumntyp. Den konverterar inte mellan familjer, så ett matchningsfel utlöser ValueError innan några rader har skrivits:

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

Använd mappningarna i Datatyp-mappningar omvänt för att välja piltypen. Money, decimala och numeriska kolumner behöver decimal128, inte float64. Data som läses tillbaka med cursor.arrow() har redan rätt datatyper, så en tabell som läses från SQL Server läses in i en motsvarande tabell utan konvertering.

Kartkolumner efter namn

När ordningen på Arrow-kolumnerna inte stämmer överens med måltabellen, ange column_mappings med namnen på måltabellens kolumner i Arrow-kolumnordning:

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

Metoden accepterar samma alternativ som cursor.bulkcopy(), inklusive batch_size, timeout, , keep_identity, table_lock, och keep_nulls. Mer information om dessa alternativ finns i Masskopiering.

Note

Att skicka en Arrow-källa till cursor.bulkcopy() utlöser TypeError och dirigerar dig till cursor.bulkcopy_arrow().

Datatypsmappningar

Arrow-hämtningsmetoderna mappar Microsoft SQL-typer till Arrow-typer på C++-nivå.

Microsoft SQL-typ Piltyp
int, smallint, tinyint, bigint int32, int16, int8, int64
flyt, reell float64, float32
decimal, numerisk decimal128
bit bool
Char, Varchar, Nchar, Nvarchar utf8
text, ntext large_utf8
binär, varbinär binary, large_binary
date date32
time time64[us]
datetime, datetime2, smalldatetime timestamp[us]
datetimeoffset timestamp[us, tz=UTC]
uniqueidentifier utf8 (sträng med versaler)
xml utf8

Note

Föraren konverterar datetimeoffset typen till UTC eftersom pilkolumner kräver en fast tidszon. Drivrutinen normaliserar tidszonsinformation per cell från Microsoft SQL till UTC under konverteringen.

Typen sql_variant stöds inte av Arrow-hämtametoder och skapar ett undantag för icke-stödd datatyp. Använd standard fetchone(), , eller fetchmany() för frågor som returnerar fetchall() kolumnersql_variant.

Prestandaöverväganden

Arrow-hämtningsmetoder är snabbast för analys och storskaliga dataoperationer, medan vanliga kursormetoder lämpar sig bättre för transaktionsmönster med små resultatmängder.

När man ska använda Arrow kontra vanlig hämtning

Scenario Rekommenderat tillvägagångssätt
Hämta några rader för visning fetchone() / fetchall()
Ladda data i Pandas eller Polars cursor.arrow()
Bearbeta stora datamängder i delar cursor.arrow_reader()
Enkelradsuppslagningar eller små resultatmängder fetchone() / fetchval()
Analys- eller aggregeringskedjor cursor.arrow() + Polars/DuckDB
Skriv resultaten i Parquet eller Arrow IPC cursor.arrow_reader() + PyArrow I/O

Minneshantering för stora datamängder

För resultatmängder som kan överstiga tillgängligt minne, använd arrow_reader() med en rimlig 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")

Volymstorlek på Tune-batchen

Parametern batch_size styr hur många rader som hämtas i varje batch. Den optimala storleken beror på din radbredd och tillgängligt minne. Bredare rader med stora kolumner som nvarchar(max) eller varbinary(max) gynnas av mindre batchstorlekar, medan smala rader gynnas av större.

  • Standard (8192): Bra balans för de flesta arbetsbelastningar.
  • Mindre (1000-5000): Används för breda tabeller med stora kolumner.
  • Större (50000-100000): Användning för smala tabeller eller när genomströmning är viktigare än minne.
# 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)

Relaterat innehåll

  • Last updated on