Użyj mssql-python z Apache Arrow

Sterownik mssql-python udostępnia metody pobierania danych w formacie Apache Arrow, umożliwiające wydajne pobieranie danych kolumnowych z Microsoft SQL i Azure SQL Database.

Apache Arrow to wielojęzyczna platforma programistyczna do przetwarzania kolumnowych danych w pamięci. Sterownik konwertuje zestawy wyników ODBC bezpośrednio na format Arrow w C++, omijając tworzenie obiektów w Python dla poprawy wydajności.

Integracja Arrow umożliwia:

• Transfer danych bez kopiowania do Polars, Pandas i DuckDB. „Zero-copy” oznacza, że dane pozostają w jednym buforze pamięci, do którego sterownik zapisuje dane, a biblioteki korzystające odczytują je bezpośrednio, więc żaden wiersz nie jest kopiowany do pośrednich obiektów Pythona.
• Strumieniowe przesyłanie zestawów wyników za pośrednictwem RecordBatchReader bez wczytywania wszystkiego do pamięci.
• Kolumnowy format danych idealny do zadań związanych z analityką i uczeniem maszynowym.
• Zmniejszone zużycie pamięci w porównaniu do tworzenia obiektów w Python wiersz po wierszu.

Metody kursora

Pakiet pyarrow jest wymagany do korzystania z metod pobierania Arrow. Zainstaluj go za pomocą polecenia pip install pyarrow. Jeśli pyarrow nie jest zainstalowany, wywołanie dowolnej metody Arrow powoduje ImportError.

Sterownik mssql-python dodaje trzy metody do obiektu kursora do dostępu do danych Arrow. Wszystkie trzy metody konwertują zestawy wyników ODBC na format Arrow w warstwie C++ sterownika, co pozwala uniknąć tworzenia pośrednich obiektów Python.

• arrow() zwraca cały zbiór wyników jako jedną tablicę w pamięci. Najprostsze w użyciu.
• arrow_batch() Zwraca jedną partię wierszy naraz, zapewniając ręczną kontrolę nad pętlą.
• arrow_reader() zwraca iterator, który automatycznie generuje partie. Najlepsze do streamowania dużych wyników.

Korzystanie z cursor.arrow(batch_size=8192)

Pobierz cały zbiór wyników jako pojedynczy pyarrow.Table. Ta metoda jest najprostsza i dobrze działa, gdy cały zbiór wyników mieści się w pamięci.

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

Korzystanie z cursor.arrow_batch(batch_size=8192)

Pobierz pojedynczy pyarrow.RecordBatch, zawierający maksymalnie batch_size wierszy. Użyj tej metody w przypadku niestandardowych pętli przetwarzania wsadowego, gdy potrzebujesz szczegółowej kontroli nad liczbą wierszy pobieranych jednocześnie.

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

Korzystanie z cursor.arrow_reader(batch_size=8192)

Zwraca czytnik, który zwraca obiekty RecordBatch do momentu wyczerpania zbioru wyników. Ta metoda jest najbardziej efektywną pod względem pamięci opcją dla dużych zbiorów wyników.

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

Czytnik przesyła wyniki za pośrednictwem połączenia, więc dopóki otwarty czytnik nie zostanie w pełni odczytany, to połączenie nie może wykonać kolejnego polecenia. Próba utworzenia elementu kończy się błędem Connection is busy with results for another command.

Trzy rzeczy uwalniają czytelnika: iteracja do końca, zamknięcie kursora nadrzędnego lub zamknięcie czytnika. Jeśli przestaniesz czytać przed wyczerpaniem zbioru wyników i nadal używasz kursora, zamknij czytnik. Jego zamknięcie resetuje również kursor macierzysty, więc można na nim uruchomić kolejną instrukcję.

Użyj czytnika jako menedżera kontekstu, tak aby zamykał się nawet wtedy, gdy wyjątek przerwie pętlę:

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

Możesz też zadzwonić bezpośrednio reader.close() . Dzwonienie więcej niż raz jest bezpieczne, a nieruchomość informuje reader.closed , czy ją zamknąłeś.

Często używane wzorce

Tabele strzałek integrują się bezpośrednio z popularnymi bibliotekami danych Python. Poniższe przykłady pokazują, jak przekazywać dane Arrow do panda, polarzy, DuckDB oraz formatów plików bez kopiowania danych.

Wczytaj wyniki do 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())

Załaduj wyniki do Polars

import polars as pl

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

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

Wyniki zapytań za pomocą DuckDB

DuckDB może zapytywać tabele strzałek bezpośrednio w SQL bez kopiowania danych. Ta możliwość jest przydatna, gdy potrzebujesz analizy w stylu SQL na zestawach wyników już w formacie Arrow.

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

Przesyłaj strumieniowo duże zbiory wyników do formatu Parquet

Dla dużych zbiorów wyników streamuj partie Arrow bezpośrednio do pliku Parquet, nie ładując całego zbioru danych do pamięci. ParquetWriter zapisuje każdą partię przyrostowo.

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

Eksport do innych formatów

PyArrow oferuje wbudowane mechanizmy zapisu do formatu CSV i formatu plików Arrow IPC (znanego również jako Feather V2). Pliki IPC Arrow zachowują dokładnie typy Arrow i są szybkie do odczytu.

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)

Załaduj dane Arrow do SQL Server

Metoda cursor.bulkcopy_arrow() zapisuje dane Arrow do tabeli bez uprzedniego konwertowania ich najpierw na krotki wierszy w Pythonie. Argument source przyjmuje dowolną z poniższych wartości:

• Element pyarrow.Table.

• Element pyarrow.RecordBatch.

• A pyarrow.RecordBatchReader, wliczając czytnik zwrócony przez cursor.arrow_reader().

• Każdy obiekt, który udostępnia interfejs danych Arrow C przez __arrow_c_stream__ lub __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")

Wartości nullowe strzałek zapisywane są jako wartości SQL NULL.

Przesyłaj zestaw wyników do innej tabeli

Ponieważ bulkcopy_arrow() akceptuje czytnik, można przenosić duży zbiór wyników między tabelami bez materializacji go w pamięci:

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

Dopasuj typy strzałek do kolumn docelowych

Writer Arrow wymaga, aby każdy typ kolumny Arrow był zgodny z docelowym typem kolumny SQL. Nie konwertuje między rodzinami, więc niedopasowanie powoduje zgłoszenie ValueError, zanim zostaną zapisane jakiekolwiek wiersze:

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

Użyj mapowań typów danych w odwrotny sposób, aby wybrać typ Arrow. Kolumny pieniężne, dziesiętne i liczbowe potrzebują decimal128, a nie float64. Dane odczytane za pomocą cursor.arrow() mają już przypisane poprawne typy, więc tabela odczytana z SQL Server jest wczytywana do odpowiadającej jej tabeli bez konwersji.

Mapuj kolumny według nazwy

Gdy kolejność kolumn Arrow nie odpowiada tabeli docelowej, przekaż column_mappings z nazwami kolumn docelowych w kolejności kolumn Arrow:

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

Metoda akceptuje te same opcje co cursor.bulkcopy(), w tym batch_size, timeout, keep_identity, table_lock, oraz keep_nulls. Więcej informacji o tych opcjach można znaleźć w Kopiowanie zbiorcze.

Mapowanie typu danych

Metody pobierania strzałek mapują typy Microsoft SQL na typy strzałek na poziomie C++.

Zagadnienia dotyczące wydajności

Metody pobierania strzałek są najszybsze do analiz i operacji z danymi zbiorczymi, natomiast standardowe metody kursora lepiej sprawdzają się w wzorcach transakcyjnych z małymi zbiorami wyników.

Kiedy używać Arrow, a kiedy standardowego fetch

Zarządzanie pamięcią dla dużych zbiorów danych

W przypadku zbiorów wyników, które mogą być większe niż dostępna pamięć, użyj arrow_reader(), ustawiając batch_size na rozsądną wartość.

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

Dostosuj rozmiar partii

Parametr ten batch_size kontroluje, ile wierszy jest pobieranych w każdej partii. Optymalny rozmiar zależy od szerokości wiersza i dostępnej pamięci. Szersze wiersze z dużymi kolumnami, takimi jak nvarchar(max) czy varbinary(max), korzystają z mniejszych rozmiarów partii, podczas gdy wąskie wiersze korzystają z większych.

• Domyślne (8192): Dobra równowaga dla większości obciążeń.
• Mniejsze (1000-5000): Używa się do szerokich tabel z dużymi kolumnami.
• Większe (50000-100000): Zastosowanie do wąskich tabel lub gdy przepustowość ma większe znaczenie niż pamięć.

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

Treści powiązane

• Użyj mssql-python z Polars
• Użyj mssql-python z pandas
• Pobieranie danych za pomocą mssql-python
• Mapowania typów danych dla mssql-python
• Dostrajanie wydajności aplikacji mssql-python