Avvio rapido: Apache Arrow con il driver mssql-python per Python

In questo quickstart, usa i metodi Arrow integrati del driver mssql-python per recuperare i dati di SQL Server come tabelle Apache Arrow in formato colonnare. Il formato di memoria colonnare di Arrow consente analisi ad alte prestazioni, interoperabilità zero-copy con pandas, Polars e DuckDB e un I/O efficiente dei file Parquet senza creare oggetti Python riga per riga.

Il mssql-python driver non richiede dipendenze esterne nei computer Windows. Il driver installa tutto ciò di cui ha bisogno con una sola pip installazione, così puoi usare l'ultima versione del driver per nuovi script senza rompere altri script che non hai tempo di aggiornare e testare.

Documentazione di mssql-python | Codice sorgente di mssql-python | Pacchetto (PyPI) | uv

Prerequisiti

  • Python 3.10 o versione successiva

  • Se Python non è già disponibile, installare il runtime Python e la gestione pacchetti pip da python.org.

  • Non si vuole usare il proprio ambiente? Segui Container e lo sviluppo locale per creare un devcontainer riproducibile o un ambiente di Codespaces su GitHub.

  • Visual Studio Code con le seguenti estensioni:

    • Estensione di Python per Visual Studio Code

  • Interfaccia Command-Line di Azure per l'autenticazione senza password in macOS e Linux.

  • Se non hai già uv, segui le istruzioni di installazione.

  • Un database su SQL Server, un database SQL di Azure o un database SQL in Fabric con lo AdventureWorks2025 schema di esempio e una stringa di connessione valida.

Installare prerequisiti specifici del sistema operativo monouso. Gli utenti Windows possono saltare questo passaggio. Per dettagli completi sulla piattaforma, vedi Installa mssql-python.

apk add libtool krb5-libs krb5-dev

Creare un database SQL

Crea o collegati a un database SQL su una delle seguenti piattaforme:

Creare il progetto ed eseguire il codice

  1. Crea un nuovo progetto
  2. Aggiungere dipendenze
  3. Avviare Visual Studio Code
  4. Aggiornare pyproject.toml
  5. Aggiornare main.py
  6. Salvare la stringa di connessione
  7. Usare uv run per eseguire lo script

Creare un nuovo progetto

  1. Apri un prompt dei comandi nella directory di sviluppo. Se non ne hai una, crea una nuova directory, come python o scripts. Evita le cartelle sul tuo OneDrive, poiché la sincronizzazione può interferire con la gestione del tuo ambiente virtuale.

  2. Crea un nuovo progetto usando uv.

    uv init arrow-qs
cd arrow-qs

Aggiungere dipendenze

Nella stessa directory, installa i pacchetti mssql-python, python-dotenv, pyarrow e rich.

uv add mssql-python python-dotenv pyarrow rich

Avviare Visual Studio Code

Nella stessa directory eseguire il comando seguente.

code .

Aggiornare pyproject.toml

  1. Il file pyproject.toml contiene i metadati del tuo progetto. Aprire il file nell'editor preferito.

  2. Esaminare il contenuto del file. Dovrebbe essere simile a questo esempio. Nota la versione Python e la dipendenza da mssql-python usare >= per definire una versione minima. Se si preferisce una versione esatta, cambiare >= prima del numero di versione in ==. Le versioni risolte di ogni pacchetto vengono quindi memorizzate nel uv.lock. Il lockfile garantisce che gli sviluppatori che lavorano sul progetto utilizzino versioni coerenti dei pacchetti. Esegui il commit di pyproject.toml e uv.lock ed esegui in CI uno scanner delle dipendenze approvato dalla tua organizzazione. Non modificare direttamente il uv.lock file.

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

  3. Aggiornare la descrizione in modo che sia più descrittiva.

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

  4. Salva e chiudi il file.

Aggiornare main.py

  1. Aprire il file denominato main.py. Dovrebbe essere simile a questo esempio.

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

if __name__ == "__main__":
    main()

  2. Sostituire l'intero contenuto di main.py con il codice seguente.

    """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:
    """Use arrow_reader() to stream results as a RecordBatchReader."""
    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()
    arrow_table = reader.read_all()
    cursor.close()
    return arrow_table


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. Use RecordBatchReader with cursor.arrow_reader()
    console.rule("[bold]cursor.arrow_reader() - RecordBatchReader[/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()

Salvare la stringa di connessione

  1. Aprire il .gitignore file e aggiungere un'esclusione per .env i file. Il file dovrebbe essere simile a questo esempio. Assicurati di salvarlo e chiuderlo quando hai finito.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

  2. Nella directory corrente creare un nuovo file denominato .env.

  3. All'interno del .env file aggiungere una voce per la stringa di connessione denominata SQL_CONNECTION_STRING. Sostituire l'esempio qui con il valore effettivo della stringa di connessione.

    SQL_CONNECTION_STRING="Server=<server_name>;Database=<database_name>;Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"

    Importante

    Mantieni .env in locale e fuori dal controllo del codice sorgente. Per gli ambienti CI e di distribuzione, inserisci la stringa di connessione o i segreti che la compongono dall'archivio segreti della piattaforma invece di copiare .env tra macchine.

    Tip

    La stringa di connessione che usi dipende in gran parte dal tipo di database SQL a cui ti stai collegando. Se ci si connette a un database SQL di Azure o a un database SQL in Fabric, usare la stringa di connessione ODBC dalla scheda Stringhe di connessione. Potrebbe essere necessario modificare il tipo di autenticazione a seconda dello scenario. Per altre informazioni sulle stringhe di connessione e sulla relativa sintassi, vedere informazioni di riferimento sulla sintassi delle stringhe di connessione.

Usare uv run per eseguire lo script

Tip

In macOS, entrambi ActiveDirectoryInteractive e ActiveDirectoryDefault funzionano per l'autenticazione di Microsoft Entra. ActiveDirectoryInteractive richiede di accedere ogni volta che si esegue lo script. Per evitare ripetuti prompt di accesso, accedi una volta tramite la CLI di interfaccia della riga di comando di Azure eseguendo az login, poi usa ActiveDirectoryDefault, che riutilizza la credenziale memorizzata nella cache.

  • Nella finestra del terminale precedente o in una nuova finestra del terminale aperta nella stessa directory eseguire il comando seguente.

    uv run main.py

    Lo script illustra tre metodi di fetch di Arrow:

    • cursor.arrow() restituisce un completo pyarrow.Table con tutte le righe. Ideale per set di risultati piccoli o medi dove serve il dataset completo in memoria.

    • cursor.arrow_batch() restituisce un pyarrow.RecordBatch alla volta. È ideale per grandi set di risultati dove vuoi elaborare i dati in modo incrementale senza caricare tutto in memoria.

    • cursor.arrow_reader() restituisce un pyarrow.RecordBatchReader per lo streaming. Ideale per l'elaborazione in stile pipeline o per il passaggio diretto alle librerie che accettano un lettore.

    Lo script salva anche i dati del prodotto in un file Parquet e li rilegge per verificare il viaggio di andata e ritorno.

Funzionamento del codice

  1. Connessione: Lo script carica la stringa di connessione da un .env file e crea una connessione usando mssql_python.connect().

  2. Recupero della tabella completa: cursor.arrow() esegue la query e restituisce l'intero insieme di risultati come un pyarrow.Table. Il driver converte i dati nel suo livello C++ utilizzando l'interfaccia dati Arrow C, bypassando la creazione di oggetti in Python per migliorare le prestazioni.

  3. Streaming in batch: cursor.arrow_batch() ne restituisce uno pyarrow.RecordBatch per chiamata. Il ciclo raccoglie blocchi finché non rimangono più righe, poi li unisce in un'unica tabella. Usa questo approccio per grandi dataset o quando vuoi elaborare ogni batch in modo indipendente.

  4. RecordBatchReader: cursor.arrow_reader() restituisce un pyarrow.RecordBatchReader, un'interfaccia Arrow standard che molte librerie accettano direttamente. La chiamata a reader.read_all() legge l'intero flusso e lo converte in una tabella.

  5. I/O Parquet: pyarrow.parquet.write_table() salva la tabella Arrow in un file Parquet compresso. Questo formato preserva i tipi di colonne e supporta letture parziali efficienti.

Passaggi successivi

Usa questi articoli per continuare a costruire:

  • Integrazione con Arrow per schemi avanzati di Arrow, tra cui l'elaborazione in batch, la gestione della memoria e l'interoperabilità con le librerie.
  • integrazione pandas per caricare direttamente i risultati delle query nei DataFrame.
  • Integrazione di Polars per creare DataFrame di Polars a partire da query native di Arrow.

Integrazione con frecce

  • Last updated on