Démarrage rapide : Apache Arrow avec le pilote mssql-python pour Python

Dans ce démarrage rapide, utilisez les mssql-python méthodes Arrow fetch intégrées au pilote pour récupérer les données de SQL Server sous forme de tables Apache Arrow en colonnes. Le format mémoire colonnaire d'Arrow permet des analyses haute performance, une interopérabilité zéro copie avec pandas, Polar et DuckDB, ainsi qu'une E/S efficace de fichiers Parquet sans création d'objets Python ligne par ligne.

Le mssql-python pilote ne nécessite aucune dépendance externe sur les machines Windows. Le pilote installe tout ce dont il a besoin en une seule pip installation, donc vous pouvez utiliser la dernière version du pilote pour de nouveaux scripts sans casser d’autres scripts que vous n’avez pas le temps de mettre à jour et de tester.

documentation mssql-python | code source mssql-python | Package (PyPI) | uv

Logiciels requis

Prérequis pour installer le système d’exploitation

Installez les prérequis ponctuels spécifiques au système d'exploitation. Les utilisateurs Windows peuvent sauter cette étape. Pour tous les détails sur la plateforme, voir Installer mssql-python.

apk add libtool krb5-libs krb5-dev libstdc++

Créer une base de données SQL

Créer ou connecter une base de données SQL sur l’une des plateformes suivantes :

Créer le projet et exécuter le code

Créer un projet

  1. Ouvrez une invite de commandes dans votre répertoire de développement. Si vous n’en avez pas, créez un nouveau répertoire, tel que python ou scripts. Évitez les dossiers sur votre OneDrive, car la synchronisation peut interférer avec la gestion de votre environnement virtuel.

  2. Créez un nouveau projet en utilisant uv.

    uv init arrow-qs
cd arrow-qs

Ajout de dépendances

Dans le même répertoire, installez les mssql-pythonpaquets , python-dotenv, pyarrow, et rich .

uv add mssql-python python-dotenv pyarrow rich

Lancer Visual Studio Code

Dans le même répertoire, exécutez la commande suivante.

code .

Mettre à jour pyproject.toml

  1. Ouvrez le fichier pyproject.toml dans votre éditeur. Ce fichier contient les métadonnées de votre projet.

  2. Passez en revue le contenu du fichier. Il doit être similaire à cet exemple. Indiquez la version de Python et la dépendance pour mssql-python ; utilisez >= pour définir une version minimale. Si vous préférez une version exacte, changez le >= avant le numéro de version en ==. Les versions résolues de chaque package sont ensuite stockées dans uv.lock. Le fichier verrouillé garantit que les développeurs travaillant sur le projet utilisent des versions cohérentes des paquets. Validez pyproject.toml et uv.lock, puis exécutez un analyseur de dépendances approuvé par l’organisation dans la CI. Ne modifie pas directement le uv.lock fichier.

    [project]
name = "arrow-qs"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "mssql-python>=1.16.0",
    "pyarrow>=19.0.0",
    "python-dotenv>=1.1.1",
    "rich>=14.1.0",
]

  3. Mettez à jour la valeur description.

    description = "Fetch SQL Server data as Apache Arrow tables using mssql-python"

  4. Enregistrez et fermez le fichier.

Mettre à jour main.py

  1. Ouvrez le fichier nommé main.py. Il doit être similaire à cet exemple.

    def main():
    print("Hello from arrow-qs!")

if __name__ == "__main__":
    main()

  2. Remplacez tout le contenu de main.py par le code suivant.

    """Fetch SQL Server data as Apache Arrow tables using mssql-python."""

from os import getenv

import pyarrow as pa
import pyarrow.parquet as pq
from dotenv import load_dotenv
from mssql_python import connect, Connection
from rich.console import Console
from rich.table import Table

console = Console()


def get_connection() -> Connection:
    """Create a connection using the connection string from .env."""
    load_dotenv()
    conn_str = getenv("SQL_CONNECTION_STRING")
    if not conn_str:
        raise ValueError("SQL_CONNECTION_STRING not set in .env file")
    return connect(conn_str)


def fetch_arrow_table(conn: Connection) -> pa.Table:
    """Run a query and return the full result as an Arrow Table."""
    cursor = conn.cursor()
    cursor.execute("""
        SELECT
            p.ProductID,
            p.Name,
            p.ProductNumber,
            p.Color,
            p.StandardCost,
            p.ListPrice,
            p.Size,
            p.Weight,
            p.SellStartDate,
            pc.Name AS Category
        FROM SalesLT.Product AS p
        INNER JOIN SalesLT.ProductCategory AS pc
            ON p.ProductCategoryID = pc.ProductCategoryID
        ORDER BY p.ListPrice DESC
    """)
    arrow_table = cursor.arrow()
    cursor.close()
    return arrow_table


def fetch_arrow_batches(conn: Connection) -> pa.Table:
    """Stream results one batch at a time using arrow_batch()."""
    cursor = conn.cursor()
    cursor.execute("""
        SELECT
            c.CustomerID,
            c.CompanyName,
            c.EmailAddress,
            COUNT(soh.SalesOrderID) AS OrderCount,
            SUM(soh.SubTotal + soh.TaxAmt + soh.Freight) AS TotalSpent
        FROM SalesLT.Customer AS c
        LEFT OUTER JOIN SalesLT.SalesOrderHeader AS soh
            ON c.CustomerID = soh.CustomerID
        GROUP BY
            c.CustomerID,
            c.CompanyName,
            c.EmailAddress
        ORDER BY TotalSpent DESC
    """)
    batches = []
    while True:
        batch = cursor.arrow_batch()
        if batch is None or batch.num_rows == 0:
            break
        batches.append(batch)
    cursor.close()

    if not batches:
        return pa.table({})

    return pa.Table.from_batches(batches)


def fetch_with_reader(conn: Connection) -> pa.Table:
    """Stream results with arrow_reader() and combine the batches."""
    cursor = conn.cursor()
    cursor.execute("""
        SELECT
            soh.SalesOrderID,
            soh.OrderDate,
            (soh.SubTotal + soh.TaxAmt + soh.Freight) AS TotalDue,
            c.CompanyName
        FROM SalesLT.SalesOrderHeader AS soh
        INNER JOIN SalesLT.Customer AS c
            ON soh.CustomerID = c.CustomerID
        ORDER BY soh.OrderDate DESC
    """)
    reader = cursor.arrow_reader()
    batches = list(reader)
    cursor.close()

    if not batches:
        return pa.table({})

    return pa.Table.from_batches(batches)


def display_arrow_table(arrow_table: pa.Table, title: str, max_rows: int = 10) -> None:
    """Display an Arrow table using rich formatting."""
    rich_table = Table(title=title)

    for name in arrow_table.column_names:
        rich_table.add_column(name, style="bright_white")

    for i in range(min(max_rows, arrow_table.num_rows)):
        row = [str(arrow_table.column(col)[i].as_py()) for col in range(arrow_table.num_columns)]
        rich_table.add_row(*row)

    if arrow_table.num_rows > max_rows:
        rich_table.add_row(*[f"... ({arrow_table.num_rows - max_rows} more rows)" if col == 0 else "" for col in range(arrow_table.num_columns)])

    console.print(rich_table)
    console.print(f"\n[dim]Schema: {arrow_table.num_columns} columns, {arrow_table.num_rows} rows[/dim]\n")


def save_to_parquet(arrow_table: pa.Table, file_path: str) -> None:
    """Save an Arrow table to a Parquet file."""
    pq.write_table(arrow_table, file_path)
    console.print(f"[green]Saved {arrow_table.num_rows} rows to {file_path}[/green]\n")


def main() -> None:
    conn = get_connection()

    # 1. Fetch entire result as an Arrow Table with cursor.arrow()
    console.rule("[bold]cursor.arrow() - Full table fetch[/bold]")
    products = fetch_arrow_table(conn)
    display_arrow_table(products, "Products (Top 10 by List Price)")

    # 2. Stream results in batches with cursor.arrow_batch()
    console.rule("[bold]cursor.arrow_batch() - Batch streaming[/bold]")
    customers = fetch_arrow_batches(conn)
    display_arrow_table(customers, "Customers by Total Spent")

    # 3. Stream results with cursor.arrow_reader()
    console.rule("[bold]cursor.arrow_reader() - Reader streaming[/bold]")
    orders = fetch_with_reader(conn)
    display_arrow_table(orders, "Recent Orders")

    # 4. Save to Parquet
    console.rule("[bold]Save to Parquet[/bold]")
    save_to_parquet(products, "products.parquet")

    # 5. Read back from Parquet and verify
    loaded = pq.read_table("products.parquet")
    console.print(f"[green]Read back {loaded.num_rows} rows from products.parquet[/green]")
    console.print(f"[dim]Schema: {loaded.schema}[/dim]\n")

    conn.close()


if __name__ == "__main__":
    main()

Enregistrer la chaîne de connexion

  1. Ouvrez le .gitignore fichier et ajoutez une exclusion pour .env les fichiers. Votre fichier devrait être similaire à l’exemple suivant.

    # Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

  2. Enregistrez et fermez .gitignore.

  3. Dans le répertoire actif, créez un fichier nommé .env.

  4. Dans le .env fichier, ajoutez une entrée pour votre chaîne de connexion nommée SQL_CONNECTION_STRING. Remplacez les espaces réservés par vos valeurs réelles.

    SQL_CONNECTION_STRING="Server=<server>;Database=<database>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"

    Important

    Gardez .env en local et hors de la gestion de version. Pour les environnements CI et déployés, injectez la chaîne de connexion ou les clés secrètes des composants depuis le magasin des secrets de votre plateforme au lieu de copier .env d’une machine à l’autre.

    La chaîne de connexion dépend du type de base de données SQL à laquelle vous vous connectez. Pour Azure SQL Database ou SQL Database dans Fabric, utilisez la chaîne de connexion ODBC depuis l’onglet des chaînes de connexion, et ajustez le type d’authentification pour votre scénario. Pour plus d’informations sur la syntaxe des chaîne de connexion, voir Chaînes de connexion pour mssql-python.

    Tip

    Sur macOS, les deux ActiveDirectoryInteractive et ActiveDirectoryDefault fonctionnent pour l’authentification Microsoft Entra. ActiveDirectoryInteractive vous invite à vous connecter chaque fois que vous exécutez le script. Pour éviter les invites de connexion répétées, connectez-vous une fois avec la commande Azure CLIaz login et utilisez ActiveDirectoryDefault. ActiveDirectoryDefault Réutilise les informations d’identification mises en cache.

Utilisez uv run pour exécuter le script

Dans la fenêtre de terminal à partir de l’avant, ou une nouvelle fenêtre de terminal ouverte au même répertoire, exécutez la commande suivante.

uv run main.py

Le script présente trois méthodes de récupération d’Arrow :

  • cursor.arrow() renvoie un pyarrow.Table complet avec toutes les lignes. Utilisez-le pour des ensembles de résultats petits à moyens où vous avez besoin de l’ensemble du jeu de données en mémoire.

  • cursor.arrow_batch() renvoie un pyarrow.RecordBatch à la fois. Utilisez-le pour de grands ensembles de résultats où vous souhaitez traiter les données de manière incrémentale sans tout charger en mémoire.

  • cursor.arrow_reader() Retourne un lecteur en streaming qui génère pyarrow.RecordBatch des objets au fur et à mesure que vous l’itérez. Utilisez-le pour le traitement de type pipeline ou pour le passage direct aux bibliothèques qui acceptent un lecteur.

Le script sauvegarde également les données produit dans un fichier Parquet et les relit pour vérifier le trajet aller-retour.

Fonctionnement du code

  1. Connexion : Le script charge la chaîne de connexion depuis un .env fichier et crée une connexion en utilisant mssql_python.connect().

  2. Recherche de table complète : cursor.arrow() exécute la requête et retourne l’ensemble des résultats sous forme de fichier pyarrow.Table. Le pilote convertit les données dans sa couche C++ en utilisant l’interface de données Arrow C, contournant la création d’objets Python pour améliorer les performances.

  3. Diffusion en série : cursor.arrow_batch() en retourne un pyarrow.RecordBatch par appel. La boucle collecte des lots jusqu’à ce qu’il ne reste plus de lignes, puis les combine en une seule table. Utilisez cette approche pour de grands ensembles de données ou lorsque vous souhaitez traiter chaque lot indépendamment.

  4. Lecteur en streaming : cursor.arrow_reader() renvoie un lecteur qui produit pyarrow.RecordBatch des objets au fur et à mesure que vous l’itérez. Par rapport à la boucle arrow_batch(), le lecteur gère son propre état d’itération, vous n’avez donc pas à vérifier vous-même si le jeu de résultats est épuisé.

    Cette fonction lit le flux jusqu’à la fin puis ferme le curseur, donc elle n’a pas besoin de fermer le lecteur. Fermez le lecteur lorsque vous arrêtez de lire tôt et continuez à utiliser le curseur. Pour plus d’informations, voir intégration Apache Arrow.

  5. E/S Parquet : pyarrow.parquet.write_table() sauvegarde la table Arrow dans un fichier Parquet compressé. Ce format préserve les types de colonnes et prend en charge des lectures partielles efficaces.

Contenu connexe

  • Dernière mise à jour :