Apache Arrow ile mssql-python kullanın

mssql-python sürücüsü, Microsoft SQL ve Azure SQL Veritabanı'den yüksek performanslı sütunlu veri alımı için Apache Arrow fetch yöntemleri sağlar.

Apache Arrow, bellek içi sütunlu veriler için diller arası bir geliştirme platformudur. Sürücü, ODBC sonuç setlerini doğrudan C++ formatında Arrow formatına dönüştürür ve performansı artırmak için Python nesne oluşturmayı atlar.

Arrow entegrasyonu şunları etkinleştiriyor:

  • Polars, Pandas ve DuckDB'ye sıfır kopya veri aktarımı. "Sıfır kopya" demek, verilerin sürücünün yazdığı tek bir bellek tamponu içinde kalması ve kütüphaneleri doğrudan tüketmesi anlamına gelir; böylece satırlar ara Python nesnelerine çoğaltılmaz.
  • Sonuç kümelerini her şeyi belleğe yüklemeden RecordBatchReader aracılığıyla akış halinde iletmek.
  • Analitik ve makine öğrenimi iş yükleri için ideal sütunlu veri formatı.
  • Satır satır Python nesnesi oluşturmaya kıyasla daha düşük bellek kullanımı.

İmleç yöntemleri

Paketin pyarrow Arrow fetch metodlarını kullanması gerekiyor. pip install pyarrow ile yükleyin. Eğer pyarrow yüklü değilse, herhangi bir Arrow yönteminin çağrılması ImportError hatasına neden olur.

mssql-python sürücüsü, Arrow veri erişimi için imleç nesnesine üç yöntem ekler. Üç yöntemin tamamı ODBC sonuç setlerini sürücünün C++ katmanında Ok formatına dönüştürür, böylece ara Python nesneleri oluşturulmasını engeller.

  • arrow() Tüm sonuç kümesini tek bir bellek içi tablo olarak döndürür. En kolay kullanım.
  • arrow_batch(), her seferinde bir satır grubu döndürerek döngüyü manuel olarak denetlemenizi sağlar.
  • arrow_reader(), otomatik olarak gruplar hâlinde öğeler üreten bir yineleyici döndürür. Büyük sonuç kümelerini akış halinde iletmek için en iyisidir.

cursor.arrow(batch_size=8192)’ı kullanma

Tüm sonuç kümesini tek bir pyarrow.Table olarak getir. Bu yöntem en basit olanıdır ve tam sonuç seti hafızaya sığdığında iyi çalışır.

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

Bağlantı dizeniz Authentication=ActiveDirectoryDefault kullanıyorsa, sürücü DefaultAzureCredential kullanır; bu da birden çok kimlik bilgisi sağlayıcısını sırayla dener. İlk bağlantı yavaş olabilir çünkü SDK çalışan bir sağlayıcı bulana kadar zincirde yürür. Üretimde, ortamınızın hangi kimlik bilgisi türünü kullandığını biliyorsanız, zincirde dolaşmayı önlemek için bunu doğrudan belirtin (örneğin, yönetilen kimlik için ActiveDirectoryMSI). Daha fazla bilgi için bkz . Microsoft Entra kimlik doğrulaması.

cursor.arrow_batch(batch_size=8192)’ı kullanma

En fazla pyarrow.RecordBatch satır içeren tek bir batch_size getirin. Bu yöntemi, aynı anda kaç satır alınacağına ince bir şekilde kontrol gerektiren özel toplu işleme döngüleri için kullanın.

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

cursor.arrow_reader(batch_size=8192)’ı kullanma

Sonuç kümesi tükenene kadar RecordBatch nesneleri üreten bir okuyucu döndürün. Bu yöntem, büyük sonuç kümeleri için en verimli bellek verimli seçenektir.

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

Okuyucu, sonuçları bağlantı üzerinden akış halinde iletir; bu nedenle, okunması tamamlanmamış bir okuyucu açıkken bu bağlantı başka bir komut başlatamaz. Birini denemek, Connection is busy with results for another command hatasıyla sonuçlanır.

Okuyucuyu serbest bırakmak için üç şey var: sona kadar yinelemek, ana imleci kapatmak ya da okuyucuyu kapatmak. Sonuç seti bitmeden okumayı bırakıp imleci kullanmaya devam ederseniz, okuyucuyu kapatın. Kapatmak ayrıca ana imleci sıfırlar, böylece üzerinde başka bir ifade çalıştırabilirsiniz.

Okuyucuyu bir bağlam yöneticisi olarak kullanın; böylece bir istisna döngüyü kesintiye uğratsa bile okuyucu kapanır:

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

Ayrıca doğrudan arayabilirsiniz reader.close() . Onu birden fazla kez çağırmak güvenlidir ve reader.closed özelliği onu kapatıp kapatmadığınızı bildirir.

Ortak desenler

Ok tabloları doğrudan popüler Python veri kütüphaneleriyle entegre olur. Aşağıdaki örnekler, Arrow verilerinin pandalar, Polars, DuckDB ve dosya formatlarına veri kopyalamadan nasıl iletileceğini gösterir.

Sonuçları pandas'a yükleyin

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

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

Sonuçları Polars'a yükle

import polars as pl

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

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

DuckDB ile sorgu sonuçları

DuckDB, veri kopyalamadan doğrudan SQL ile Arrow tablolarını sorgulayabilir. Bu özellik, zaten Arrow formatında olan sonuç kümelerinde SQL tarzı analize ihtiyacınız olduğunda faydalıdır.

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

Büyük sonuç kümelerini Parquet’e akış olarak aktarın

Büyük sonuç kümeleri için, tüm veri kümesini belleğe yüklemeden Arrow batch’lerini doğrudan bir Parquet dosyasına aktarın. ParquetWriter her partiyi aşamalı olarak yazar.

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

Diğer formatlara ihracat

PyArrow, CSV ve Arrow IPC dosya formatı (Feather V2 olarak da bilinir) için yerleşik yazıcılar sağlar. Arrow IPC dosyaları, Arrow veri türlerini tam olarak korur ve yeniden hızlıca okunabilir.

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)

Arrow verilerini SQL Server'a yükle

Yöntem, cursor.bulkcopy_arrow() Arrow verilerini önce Python satır tuple'larına dönüştürmeden bir tabloya yazar. Argüman source aşağıdakilerden herhangi birini kabul eder:

  • Bir pyarrow.Table.

  • Bir pyarrow.RecordBatch.

  • Bir pyarrow.RecordBatchReader, cursor.arrow_reader() tarafından döndürülen okuyucu dahil.

  • Arrow C veri arayüzünü __arrow_c_stream__ veya __arrow_c_array__ aracılığıyla sunan herhangi bir nesne.

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

Arrow null değerleri, SQL NULL değerleri olarak yazılır.

Bir sonuç kümesini başka bir tabloya aktarın

bulkcopy_arrow(), bir okuyucuyu kabul ettiğinden, büyük bir sonuç kümesini bellekte oluşturmadan tablolar arasında taşıyabilirsiniz:

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

Ok tiplerini hedef sütunlara eşleştirin

Arrow yazarı, her Arrow sütun tipinin hedef SQL sütun tipiyle uyumlu olmasını gerektirir. Aileler arasında dönüşüm yapmaz, bu yüzden satırlar yazılmadan önce uyumsuzluk ortaya çıkar ValueError :

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

Ok tipini seçmek için Veri tipi eşlemelerindeki eşlemeleri ters olarak kullanın. para, ondalık ve sayısal sütunlar decimal128 gerektirir, float64 değil. cursor.arrow() ile geri okunan veriler zaten doğru türleri içerdiğinden, SQL Server'dan okunan bir tablo dönüştürme gerektirmeden karşılık gelen bir tabloya yüklenir.

Harita sütunları isimlere göre

Ok sütun sırası hedef tablo ile eşleşmiyorsa, column_mappings hedef sütun isimlerini Ok sütun sırasıyla aktarın:

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

Yöntem, cursor.bulkcopy() ile aynı seçenekleri kabul eder; bunlara batch_size, keep_identity, timeout, table_lock ve keep_nulls dahildir. Bu seçenekler hakkında daha fazla bilgi için Toplu kopya sayfasına bakınız.

Note

Bir Arrow kaynağını cursor.bulkcopy() içine geçirmek, TypeError hatasına neden olur ve sizi cursor.bulkcopy_arrow() konumuna yönlendirir.

Veri türü eşlemeleri

Arrow fetch yöntemleri, Microsoft SQL tiplerini C++ seviyesinde Arrow tipleriyle eşler.

Microsoft SQL tipi Ok türü
int, smallint, tinyint, bigint int32, int16, int8, int64
float, reel float64, float32
ondalık, sayısal decimal128
bit bool
char, varchar, nchar, nvarchar utf8
metin, nmetin large_utf8
binary, varbinary binary, large_binary
date date32
time time64[us]
datetime, datetime2, smalldatetime timestamp[us]
datetimeoffset timestamp[us, tz=UTC]
uniqueidentifier utf8 (büyük harflerden oluşan dize)
xml utf8

Note

Sürücü, Ok sütunlarının sabit bir zaman dilimi gerektirdiği için tipi datetimeoffset UTC'ye dönüştürür. Sürücü, dönüşüm sırasında Microsoft SQL'den UTC'ye hücre başına zaman dilimi bilgisini normalleştirir.

Bu sql_variant tür Arrow fetch metodları tarafından desteklenmez ve desteklenmeyen bir veri tipi istisnası oluşturur. fetchone() sütun döndüren sorgular için standart fetchmany(), fetchall() veya sql_variant kullanın.

Performansla ilgili dikkat edilmesi gerekenler

Ok getirme yöntemleri analitik ve toplu veri işlemleri için en hızlısıdır, standart imleç yöntemleri ise küçük sonuç setlerine sahip işlemsel kalıplar için daha uygundur.

Arrow yerine standart fetch ne zaman kullanılmalı?

Scenario Önerilen yaklaşım
Görüntülemek için birkaç satır getir fetchone() / fetchall()
Verileri pandalara veya Polarlara yükleyin cursor.arrow()
Büyük veri kümelerini parçalar halinde işlemek cursor.arrow_reader()
Tek satırlı aramalar veya küçük sonuç kümeleri fetchone() / fetchval()
Analitik veya toplama boru hatları cursor.arrow() + Polars/DuckDB
Sonuçları Parquet veya Arrow IPC'ye yazın cursor.arrow_reader() + PyArrow I/O

Büyük veri setleri için bellek yönetimi

Kullanılabilir belleği aşabilecek sonuç kümeleri için, arrow_reader() öğesini makul bir batch_size ile kullanın.

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

Toplu işlem boyutunu ayarlayın

batch_size parametresi, her toplu işlemde kaç satır alınacağını kontrol eder. Optimal boyut, sıra genişliğinize ve mevcut belleğe bağlıdır. Nvarchar(max) veya varbinary(max) gibi büyük sütunlu daha geniş sıralar daha küçük parti boyutlarından faydalanırken, dar sıralar daha büyük partilerden faydalanır.

  • Varsayılan (8192): Çoğu iş yükü için iyi bir denge.
  • Daha küçük (1000-5000): Geniş sütunlu geniş masalar için kullanılır.
  • Daha Büyük (50000-100000): Dar tablolar için veya aktarım kapasitesinin bellekten daha önemli olduğu durumlarda kullanılır.
# 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)

İlgili içerik

  • Last updated on