Utilisez mssql-python avec Apache Arrow

Le pilote mssql-python fournit des méthodes de récupération Apache Arrow pour la récupération de données columnaires haute performance depuis Microsoft SQL et Azure SQL Database.

Apache Arrow est une plateforme de développement multi-langages pour les données en colonnes en mémoire. Le pilote convertit directement les ensembles de résultats ODBC en format Arrow en C++, contournant la création d’objets Python pour améliorer les performances.

L’intégration de Arrow permet :

• Transfert de données sans copie vers Polars, pandas et DuckDB. « Zero-copy » signifie que les données restent dans un seul tampon mémoire que le pilote écrit et que les bibliothèques consommantes lisent directement, de sorte qu’aucune ligne n’est dupliquée en objets Python intermédiaires.
• Le résultat de streaming passe sur RecordBatchReader sans tout charger en mémoire.
• Format de données en colonnes idéal pour les charges de travail analytiques et d’apprentissage automatique.
• Réduction de la consommation de mémoire par rapport à la création d’objets Python ligne par ligne.

Méthodes de Curseur

Le pyarrow package doit utiliser des méthodes de récupération Arrow. Installez-le avec pip install pyarrow. Si pyarrow n’est pas installé, appeler n’importe quelle méthode Arrow génère un ImportError.

Le pilote mssql-python ajoute trois méthodes à l’objet curseur pour l’accès aux données Arrow. Les trois méthodes convertissent les ensembles de résultats ODBC au format Arrow dans la couche C++ du pilote, ce qui évite de créer des objets Python intermédiaires.

• arrow() renvoie l’ensemble des résultats sous forme d’une seule table en mémoire. Les plus simples à utiliser.
• arrow_batch() renvoie un lot de lignes à la fois, vous donnant le contrôle manuel de la boucle.
• arrow_reader() renvoie un itérateur qui génère automatiquement des lots. Idéal pour diffuser de gros résultats.

Utilisation de cursor.arrow(batch_size=8192)

Récupérez l’ensemble des résultats comme un seul pyarrow.Table. Cette méthode est la plus simple et fonctionne bien lorsque l’ensemble complet des résultats tient en mémoire.

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

Utilisation de cursor.arrow_batch(batch_size=8192)

Récupérez un seul pyarrow.RecordBatch contenant jusqu’à batch_size lignes. Utilisez cette méthode pour des boucles de traitement batch personnalisées où vous avez besoin d’un contrôle précis sur le nombre de lignes récupérées à la fois.

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

Utilisation de cursor.arrow_reader(batch_size=8192)

Retournez un lecteur qui renvoie des objets RecordBatch jusqu’à épuisement du jeu de résultats. Cette méthode est l’option la plus efficace en mémoire pour les grands ensembles de résultats.

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

Le lecteur diffuse les résultats via la connexion, donc tant qu’un lecteur non lu est ouvert, cette connexion ne peut pas lancer une autre déclaration. La tentative d’en exécuter une échoue avec une erreur Connection is busy with results for another command.

Trois choses libèrent le lecteur : l’itérer jusqu’à la fin, fermer le curseur parent ou fermer le lecteur. Si vous arrêtez de lire avant que le jeu de résultats ne soit épuisé et que vous continuez à utiliser le curseur, fermez le lector. La fermer réinitialise aussi le curseur parent, donc vous pouvez exécuter une autre instruction dessus.

Utilisez le lecteur comme gestionnaire de contexte afin qu’il se ferme même si une exception interrompt la boucle :

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

Vous pouvez aussi appeler reader.close() directement. L’appeler plusieurs fois ne présente aucun risque, et la propriété reader.closed indique si vous l’avez fermée.

Modèles courants

Les tables de flèches s’intègrent directement avec les bibliothèques de données Python populaires. Les exemples suivants montrent comment transmettre les données Arrow aux pandas, Polars, DuckDB et formats de fichiers sans copier les données.

Charger les résultats dans 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())

Charger les résultats dans Polars

import polars as pl

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

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

Résultats de requêtes avec DuckDB

DuckDB peut interroger directement les tables de flèches en SQL sans copier les données. Cette fonctionnalité est utile lorsque vous avez besoin d’une analyse de type SQL sur des ensembles de résultats déjà au format 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())

Exporter en flux des jeux de résultats volumineux vers Parquet

Pour de grands ensembles de résultats, diffusez des lots Arrow directement vers un fichier Parquet sans charger l’ensemble de données en mémoire. Le ParquetWriter écrit chaque lot de façon incrémentale.

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

Exportation vers d’autres formats

PyArrow fournit des graveurs intégrés pour CSV et le format de fichier IPC Arrow (également connu sous le nom de Feather V2). Les fichiers Arrow IPC conservent exactement les types Arrow et peuvent être relus rapidement.

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)

Charger les données Arrow dans SQL Server

La méthode cursor.bulkcopy_arrow() inscrit les données Arrow dans une table sans d’abord les convertir en tuples de lignes Python. L’argument source accepte l’une des conditions suivantes :

• Un pyarrow.Table.

• Un pyarrow.RecordBatch.

• A pyarrow.RecordBatchReader, y compris le lecteur renvoyé par cursor.arrow_reader().

• Tout objet qui expose l’interface de données Arrow C via __arrow_c_stream__ ou __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")

Les valeurs nulles des flèches sont écrites sous forme de valeurs SQL NULL.

Diffusez un ensemble de résultats dans une autre table

Parce qu’accepte bulkcopy_arrow() un lecteur, vous pouvez déplacer un grand ensemble de résultats entre tables sans le matérialiser en mémoire :

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

Associer les types de flèches aux colonnes de destination

Le module d’écriture Arrow nécessite que chaque type de colonne Arrow soit compatible avec le type de colonne SQL de destination correspondant. Il n’effectue pas de conversion d’une famille à l’autre, donc une incompatibilité déclenche ValueError avant l’écriture de la moindre ligne :

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

Utilisez la correspondance des types de données dans Correspondance des types de données en sens inverse pour choisir le type Arrow. les colonnes argent, décimales et numériques ont besoin decimal128, non float64. Les données relues avec cursor.arrow() ont déjà les types corrects ; ainsi, une table lue depuis SQL Server est chargée dans une table correspondante sans conversion.

Colonnes de la carte par nom

Lorsque l’ordre des colonnes Arrow ne correspond pas à la table de destination, passez column_mappings avec les noms des colonnes de destination dans l’ordre des colonnes 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"],
)

La méthode accepte les mêmes options que cursor.bulkcopy(), y compris batch_size, timeout, keep_identity, table_lock, et keep_nulls. Pour plus d’informations sur ces options, voir Copie en masse.

Mappages de types de données

Les méthodes de récupération Arrow associent les types SQL Microsoft aux types Arrow au niveau C++.

Considérations relatives aux performances

Les méthodes de récupération de flèches sont les plus rapides pour l’analytique et les opérations de données en masse, tandis que les méthodes de curseur standard conviennent mieux aux modèles transactionnels avec de petits jeux de résultats.

Quand utiliser Arrow ou la récupération standard

Gestion de la mémoire pour de grands ensembles de données

Pour les jeux de résultats susceptibles de dépasser la mémoire disponible, utilisez arrow_reader() avec un batch_size raisonnable.

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

Ajuster la taille du lot

Le batch_size paramètre contrôle combien de lignes sont récupérées dans chaque lot. La taille optimale dépend de la largeur de votre ligne et de la mémoire disponible. Les rangées plus larges avec de grandes colonnes comme nvarchar(max) ou varbinary(max) bénéficient de tailles de lots plus petites, tandis que les rangées étroites bénéficient de plus grandes.

• Par défaut (8192) : Bon équilibre pour la plupart des charges de travail.
• Plus petits (1000-5000) : À utiliser pour de larges tables avec de grandes colonnes.
• Plus grand (50000-100000) : Utilisation pour des tables étroites ou lorsque le débit compte plus que la mémoire.

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

Contenu connexe

• Utiliser mssql-python avec Polars
• Utilisez mssql-python avec pandas
• Récupérer des données avec mssql-python
• Mappage de types de données pour mssql-python
• Réglage des performances pour les applications mssql-python