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

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

  1. Een nieuw project maken
  2. Afhankelijkheden toevoegen
  3. Visual Studio Code starten
  4. Pyproject.toml bijwerken
  5. Main.py bijwerken
  6. De verbindingsreeks opslaan
  7. Uv-uitvoering gebruiken om het script uit te voeren

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. Het pyproject.toml-bestand bevat de metadata van je project. Open het bestand in uw favoriete editor.

  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.5.0",
    "pyarrow>=19.0.0",
    "python-dotenv>=1.1.1",
    "rich>=14.1.0",
]

  3. Werk de beschrijving bij zodat deze meer beschrijvend is.

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

De verbindingsreeks opslaan

  1. Open het .gitignore bestand en voeg een uitsluiting toe voor .env bestanden. Het bestand moet er ongeveer uitzien als in dit voorbeeld. Zorg ervoor dat u deze opslaat en sluit wanneer u klaar bent.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

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

  3. Voeg in het .env bestand een vermelding toe voor de verbindingsreeks met de naam SQL_CONNECTION_STRING. Vervang het voorbeeld hier door de werkelijke verbindingsreekswaarde.

    SQL_CONNECTION_STRING="Server=<server_name>;Database=<database_name>;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.

    Tip

    De verbindingsreeks die je gebruikt hangt grotendeels af van het type SQL-database waarmee je verbinding maakt. Als u verbinding maakt met een Azure SQL Database of een SQL-database in Fabric, gebruikt u de ODBC-verbindingsreeks op het tabblad Verbindingsreeksen. Mogelijk moet u het verificatietype aanpassen, afhankelijk van uw scenario. Zie de naslaginformatie over de syntaxis van de verbindingsreeks voor meer informatie over verbindingsreeksen en de bijbehorende syntaxis.

Gebruik uv run om het script uit te voeren

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 aanmeldingsprompts te voorkomen, meld je je één keer aan via de Azure CLI door az login uit te voeren en gebruik vervolgens ActiveDirectoryDefault, waarmee de in de cache opgeslagen referentiegegevens opnieuw worden gebruikt.

  • 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. Het beste 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. Het beste voor grote resultatensets waarbij je data incrementeel wilt verwerken zonder alles in het geheugen te laden.

    • cursor.arrow_reader() geeft een pyarrow.RecordBatchReader terug voor streaming. Het meest geschikt voor pijplijnverwerking of om rechtstreeks door te geven aan bibliotheken die een reader ondersteunen.

    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. RecordBatchReader: cursor.arrow_reader() levert een pyarrow.RecordBatchReader, een standaard Arrow-interface die door veel bibliotheken direct wordt geaccepteerd. Het aanroepen van reader.read_all() leest de volledige stream in in een tabel.

  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.

Volgende stappen 

Gebruik deze artikelen om verder te bouwen:

  • Arrow-integratie voor geavanceerde Arrow-patronen, waaronder batchverwerking, geheugenbeheer en bibliotheekinteroperabiliteit.
  • pandas-integratie om queryresultaten direct in DataFrames te laden.
  • Polars-integratie om Polars DataFrames op te bouwen uit Arrow-native query's.

Pijlintegratie

  • Last updated on