Quickstart: Apache Arrow met de mssql-python-driver voor Python

In deze quickstart gebruik je de ingebouwde Arrow-fetch-methoden van de mssql-python driver om SQL Server-gegevens op te halen als kolomvormige Apache Arrow-tabellen. Arrow's kolomgeoriënteerde geheugenindeling maakt hoogwaardige analyses, zero-copy-interoperabiliteit met pandas, Polars en DuckDB, en efficiënte Parquet-bestands-I/O mogelijk zonder Python-objecten rij voor rij aan te maken.

Het mssql-python stuurprogramma vereist geen externe afhankelijkheden op Windows-computers. De driver installeert alles wat hij nodig heeft met één enkele pip installatie, zodat je de nieuwste versie van de driver kunt gebruiken voor nieuwe scripts zonder andere scripts te breken die je niet kunt upgraden en testen.

documentatie | mssql-python-broncode | Pakket (PyPI) | Uv

Prerequisites

Installeer vereisten voor het besturingssysteem

Installeer eenmalige vereisten voor het besturingssysteem. Windows-gebruikers kunnen deze stap overslaan. Voor volledige platformdetails, zie Install mssql-python.

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

Een SQL-database maken

Maak een SQL-database aan of maak verbinding met een van de volgende platforms:

Het project maken en de code uitvoeren

Een nieuw project maken

  1. Open een opdrachtprompt in uw ontwikkelingsmap. Als je er geen hebt, maak dan een nieuwe map aan, zoals python of scripts. Vermijd mappen op je OneDrive, want synchronisatie kan het beheer van je virtuele omgeving verstoren.

  2. Maak een nieuw project aan door gebruik te maken van uv.

    uv init arrow-qs
cd arrow-qs

Afhankelijkheden toevoegen

Installeer in dezelfde map de mssql-python, python-dotenv, pyarrow, en rich pakketten.

uv add mssql-python python-dotenv pyarrow rich

Visual Studio Code starten

Voer in dezelfde map de volgende opdracht uit.

code .

Pyproject.toml bijwerken

  1. Open het pyproject.toml bestand in de editor. Dit bestand bevat de metadata van je project.

  2. Controleer de inhoud van het bestand. Deze moet vergelijkbaar zijn met dit voorbeeld. Let op de Python-versie en afhankelijkheid voor mssql-python; gebruik >= om een minimumversie te definiëren. Als u de voorkeur geeft aan een exacte versie, wijzigt u de >= voor het versienummer in ==. De opgeloste versies van elk pakket worden vervolgens opgeslagen in de uv.lock. Het lockfile zorgt ervoor dat ontwikkelaars die aan het project werken consistente pakketversies gebruiken. Commet zowel pyproject.toml als uv.lock, en voer een door de organisatie goedgekeurde afhankelijkheidsscanner uit in CI. Bewerk het uv.lock bestand niet direct.

    [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. Werk de description waarde bij.

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

  4. Sla het bestand op en sluit het.

Main.py bijwerken

  1. Open het bestand met de naam main.py. Deze moet vergelijkbaar zijn met dit voorbeeld.

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

if __name__ == "__main__":
    main()

  2. Vervang de volledige inhoud van main.py door de volgende code.

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

De verbindingsreeks opslaan

  1. Open het .gitignore bestand en voeg een uitsluiting toe voor .env bestanden. Je dossier zou vergelijkbaar moeten zijn met het volgende voorbeeld.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

  2. Opslaan en sluiten .gitignore.

  3. Maak in de huidige map een nieuw bestand met de naam .env.

  4. Voeg in het .env bestand een vermelding toe voor de verbindingsreeks met de naam SQL_CONNECTION_STRING. Vervang de tijdelijke aanduidingen door de werkelijke waarden.

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

    Important

    Blijf .env lokaal en buiten bronbeheer. Voor CI en gedeployte omgevingen injecteer je de verbindingsreeks of de componentgeheimen ervan uit je platformgeheime opslag in plaats van te kopiëren .env tussen machines.

    De verbindingsreeks hangt af van het type SQL-database waarmee je verbinding maakt. Voor Azure SQL Database of SQL Database in Fabric gebruik je de ODBC-verbindingsreeks van het tabblad verbindingsreekss en pas je het authenticatietype aan voor jouw situatie. Voor meer informatie over de syntaxis van verbindingsreeks, zie Connection strings voor mssql-python.

    Tip

    In macOS werken beide ActiveDirectoryInteractive en ActiveDirectoryDefault voor Microsoft Entra-verificatie. ActiveDirectoryInteractive u wordt gevraagd u aan te melden telkens wanneer u het script uitvoert. Om herhaalde aanmeldprompts te vermijden, log je één keer in met het Azure CLI-commandoaz login en gebruik ActiveDirectoryDefault. ActiveDirectoryDefault Hergebruikt de gecachte inloggegevens.

Gebruik uv run om het script uit te voeren

Voer in het terminalvenster van vóór, of een nieuw terminalvenster dat is geopend in dezelfde map, de volgende opdracht uit.

uv run main.py

Het script demonstreert drie Arrow-ophaalmethoden:

  • cursor.arrow() geeft een complete pyarrow.Table terug met alle rijen. Gebruik het voor kleine tot middelgrote resultaatsets waarbij je de volledige dataset in het geheugen nodig hebt.

  • cursor.arrow_batch() geeft pyarrow.RecordBatch één voor één terug. Gebruik het voor grote resultatensets waarbij je data incrementeel wilt verwerken zonder alles in het geheugen te laden.

  • cursor.arrow_reader() geeft een streaming-lezer terug die pyarrow.RecordBatch-objecten oplevert terwijl je erdoorheen itereert. Gebruik het voor pipeline-achtige verwerking of direct doorgeven aan bibliotheken die een lezer accepteren.

Het script slaat de productgegevens ook op in een Parquet-bestand en leest deze terug om de retour te verifiëren.

Hoe de code werkt

  1. Connection: Het script laadt de verbindingsreeks uit een .env bestand en maakt een verbinding aan met mssql_python.connect().

  2. Full table fetch: cursor.arrow() voert de query uit en retourneert de volledige resultaatset als een pyarrow.Table. De driver converteert data in zijn C++-laag via de Arrow C Data Interface, waarbij Python-objectcreatie wordt omzeild voor betere prestaties.

  3. Batch-streaming: cursor.arrow_batch() retourneert bij elke oproep één pyarrow.RecordBatch. De lus verzamelt batches totdat er geen rijen meer over zijn, en combineert ze vervolgens tot één tabel. Gebruik deze aanpak voor grote datasets of wanneer je elke batch onafhankelijk wilt verwerken.

  4. Streaming reader: cursor.arrow_reader() geeft een lezer terug die pyarrow.RecordBatch-objecten oplevert terwijl je erover itereert. In vergelijking met de arrow_batch() lus beheert de lezer zijn eigen iteratietoestand, zodat je zelf niet test voor een uitgeputte resultaatset.

    Deze functie leest de stream volledig en sluit vervolgens de cursor, zodat hij de lezer niet hoeft te sluiten. Sluit de lezer als je vroegtijdig stopt met lezen en blijf de cursor gebruiken. Voor meer informatie, zie Apache Arrow-integratie.

  5. Parquet I/O: pyarrow.parquet.write_table() slaat de Arrow-tabel op in een gecomprimeerd Parquet-bestand. Dit formaat behoudt kolomtypen en ondersteunt efficiënte gedeeltelijke leesopdrachten.

Verwante onderwerpen

  • Laatst bijgewerkt om