Rychlý start: Apache Arrow s ovladačem mssql-python pro Python

V tomto rychlém spuštění použijte vestavěné metody načítání Arrow ovladače mssql-python k získání dat ze SQL Server ve formě sloupcových tabulek Apache Arrow. Sloupcový formát paměti Arrow umožňuje vysoce výkonnou analytiku, interoperaci bez kopírování s pandas, Polars a DuckDB a efektivní I/O souborů Parquet bez vytváření Python objektů po řádcích.

Ovladač mssql-python nevyžaduje žádné externí závislosti na počítačích s Windows. Ovladač nainstaluje vše, co potřebuje, jednou pip instalací, takže můžete použít nejnovější verzi ovladače pro nové skripty, aniž byste rozbili jiné skripty, které nemáte čas aktualizovat a testovat.

Dokumentace mssql-python, zdrojový kód mssql-python, Balíček (PyPI) uv

Předpoklady

  • Python 3.10 nebo novější

  • Pokud ještě nemáte Python, nainstalujte Python runtime a správce balíčků pip z python.org.

  • Nechcete používat vlastní prostředí? Postupujte podle Kontejner a místní vývoj a vytvořte reprodukovatelné vývojové prostředí devcontainer nebo GitHub Codespaces.

  • Visual Studio Code s následujícími rozšířeními:

  • Azure Command-Line Interface (CLI) pro ověřování bez hesla v systému macOS a Linux.

  • Pokud ještě uvnemáte, postupujte podle pokynů k instalaci.

  • Databáze na SQL Serveru, Azure SQL Database nebo SQL databáze v Fabric s ukázkovým AdventureWorks2025 schématem a platným připojovacím řetězcem.

Nainstalujte požadavky specifické pro jednorázový operační systém. Uživatelé Windows mohou tento krok přeskočit. Pro úplné podrobnosti o platformě viz Instalace mssql-python.

apk add libtool krb5-libs krb5-dev

Vytvoření databáze SQL

Vytvořte nebo se připojte k SQL databázi na jedné z následujících platforem:

Vytvoření projektu a spuštění kódu

  1. Vytvoření nového projektu
  2. Přidání závislostí
  3. Spuštění editoru Visual Studio Code
  4. Aktualizace pyproject.toml
  5. Aktualizace main.py
  6. Ulož připojovací řetězec
  7. Použij uv run pro spuštění skriptu

Vytvoření nového projektu

  1. Otevřete příkazový řádek ve vývojovém adresáři. Pokud žádný nemáte, vytvořte nový adresář, například python nebo scripts. Vyhněte se složkám na OneDrive, protože synchronizace může narušit správu virtuálního prostředí.

  2. Vytvořte nový projekt pomocí uv.

    uv init arrow-qs
cd arrow-qs

Přidejte závislosti

Do stejného adresáře nainstalujte balíčky mssql-python, python-dotenv, pyarrow, a rich balíčky.

uv add mssql-python python-dotenv pyarrow rich

Spusťte Visual Studio Code

Ve stejném adresáři spusťte následující příkaz.

code .

Aktualizace pyproject.toml

  1. Soubor pyproject.toml obsahuje metadata vašeho projektu. Otevřete soubor v oblíbeném editoru.

  2. Zkontrolujte obsah souboru. Měl by se podobat tomuto příkladu. Všimněte si verze Pythonu a závislosti pro mssql-python; k definování minimální verze použijte >=. Pokud dáváte přednost přesné verzi, změňte >= před číslem verze na ==. Vyřešené verze každého balíčku jsou pak uloženy v uv.lock. Zámek zajišťuje, že vývojáři pracující na projektu používají konzistentní verze balíčků. Potvrďte oba pyproject.toml a uv.lock a v CI spusťte skener závislostí schválený organizací. Neupravujte uv.lock soubor přímo.

    [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. Aktualizujte popis, aby byl popisnější.

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

  4. Uložte a zavřete soubor.

Aktualizace main.py

  1. Otevřete soubor s názvem main.py. Měl by se podobat tomuto příkladu.

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

if __name__ == "__main__":
    main()

  2. Celý obsah main.py nahraďte následujícím kódem.

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

Uložte řetězec připojení

  1. .gitignore Otevřete soubor a přidejte vyloučení souborů.env. Soubor by měl být podobný tomuto příkladu. Až budete hotovi, nezapomeňte ho uložit a zavřít.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

  2. V aktuálním adresáři vytvořte nový soubor s názvem .env.

  3. .env V souboru přidejte položku pro připojovací řetězec s názvem SQL_CONNECTION_STRING. Příklad zde nahraďte skutečnou hodnotou připojovacího řetězce.

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

    Important

    Držte se .env lokálně a mimo kontrolu zdrojového kódu. Pro CI a nasazená prostředí vkládejte připojovací řetězec nebo jeho komponentní tajemství z platformy secret store místo kopírování .env mezi stroji.

    Tip

    Použitý připojovací řetězec závisí do značné míry na typu SQL databáze, ke které se připojuješ. Pokud se připojujete k Azure SQL Database nebo k SQL databázi ve Fabric, použijte připojovací řetězec ODBC z karty Připojovací řetězce. Možná budete muset upravit typ ověřování v závislosti na vašem scénáři. Další informace o připojovacích řetězcích a jejich syntaxi najdete v referenčních informacích k syntaxi připojovacího řetězce.

Použijte 'uv run' ke spuštění skriptu

Tip

V systému macOS funguje ověřování pomocí Microsoft Entra jak s ActiveDirectoryInteractive, tak s ActiveDirectoryDefault. ActiveDirectoryInteractive vás vyzve k přihlášení při každém spuštění skriptu. Abychom se vyhnuli opakovaným výzvám k přihlášení, přihlaste se jednou přes Azure CLI spuštěním az login, poté použijte ActiveDirectoryDefault, které znovu použije cacheované přihlašovací údaje.

  • V okně terminálu před nebo v novém okně terminálu, které se otevře ve stejném adresáři, spusťte následující příkaz.

    uv run main.py

    Skript ukazuje tři metody načítání Arrow:

    • cursor.arrow() vrátí kompletní pyarrow.Table se všemi řádky. Nejlepší pro malé až střední sady výsledků, kde potřebujete celý dataset v paměti.

    • cursor.arrow_batch() vrací vždy jedno pyarrow.RecordBatch. Nejlepší pro velké sady výsledků, kde chcete zpracovávat data postupně, aniž byste vše načítali do paměti.

    • cursor.arrow_reader() vrací pyarrow.RecordBatchReader pro streamování. Nejlepší pro zpracování ve stylu pipeline nebo přímé předávání knihovnám, které přijímají čtečku.

    Skript také ukládá produktová data do souboru Parquet a znovu je čte pro ověření okružní cesty.

Jak kód funguje

  1. Connection: Skript načte připojovací řetězec ze .env souboru a vytvoří spojení pomocí mssql_python.connect().

  2. Načtení celé tabulky: cursor.arrow() provede dotaz a vrátí celou množinu výsledků jako pyarrow.Table. Ovladač převádí data ve své C++ vrstvě pomocí datového rozhraní Arrow C, čímž obchází tvorbu objektů v Python pro lepší výkon.

  3. Dávkové streamování: cursor.arrow_batch() vrací při každém volání jeden pyarrow.RecordBatch. Smyčka sbírá dávky, dokud nezůstanou žádné další řádky, poté je spojí do jedné tabulky. Tento přístup použijte pro velké datové sady nebo když chcete zpracovávat každou dávku samostatně.

  4. RecordBatchReader: cursor.arrow_reader() vrací pyarrow.RecordBatchReader, standardní rozhraní Arrow, které mnoho knihoven přijímá přímo. Volání reader.read_all() načte celý proud do podoby tabulky.

  5. Parquet I/O: pyarrow.parquet.write_table() ukládá tabulku Arrow do komprimovaného souboru Parquet. Tento formát zachovává typy sloupců a podporuje efektivní částečné čtení.

Další kroky

Použijte tyto články k dalšímu budování:

  • Integrace s Arrow pro pokročilé vzory Arrow včetně dávkového zpracování, správy paměti a interoperability knihoven.
  • integrace pandas pro načítání výsledků dotazů přímo do DataFrames.
  • Integrace s Polars pro vytváření datových rámců Polars z dotazů nativně podporovaných Arrow.

Šipková integrace

  • Last updated on