Quickstart: Apache Arrow mit dem mssql-python-Treiber für Python

In diesem Quickstart verwenden Sie die integrierten Arrow-Fetch-Methoden des mssql-python Treibers, um SQL Server-Daten als spaltenartige Apache-Pfeiltabellen abzurufen. Arrows spaltenorientiertes Speicherformat ermöglicht leistungsstarke Analysen, Zero-Copy-Interoperabilität mit Pandas, Polars und DuckDB sowie eine effiziente Ein- und Ausgabe von Parquet-Dateien, ohne zeilenweise Python-Objekte erzeugen zu müssen.

Der mssql-python Treiber erfordert keine externen Abhängigkeiten von Windows-Computern. Der Treiber installiert alles, was er braucht, mit einer einzigen pip Installation, sodass du die neueste Version des Treibers für neue Skripte verwenden kannst, ohne andere Skripte zu brechen, die du nicht aktualisieren und testen kannst.

mssql-python-Dokumentation | mssql-python-Quellcode | Paket (PyPI) | Uv

Voraussetzungen

Betriebssystem-Voraussetzungen installieren

Installieren Sie die einmaligen betriebsystem-spezifischen Voraussetzungen. Windows-Nutzer können diesen Schritt überspringen. Für vollständige Plattformdetails siehe Install mssql-python.

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

Erstellen einer SQL-Datenbank

Erstellen oder verbinden Sie sich mit einer SQL-Datenbank auf einer der folgenden Plattformen:

Erstellen Sie das Projekt, und führen Sie den Code aus.

Neues Projekt erstellen

  1. Öffnen Sie eine Eingabeaufforderung in Ihrem Entwicklungsverzeichnis. Wenn du keines hast, erstelle ein neues Verzeichnis, wie zum Beispiel python oder scripts. Vermeide Ordner auf deinem OneDrive, da Synchronisation die Verwaltung deiner virtuellen Umgebung beeinträchtigen kann.

  2. Erstelle ein neues Projekt mit uv.

    uv init arrow-qs
cd arrow-qs

Hinzufügen von Abhängigkeiten

Im selben Verzeichnis installieren Sie die mssql-python, python-dotenv, pyarrow, und rich Pakete.

uv add mssql-python python-dotenv pyarrow rich

Starten Sie Visual Studio Code.

Führen Sie im selben Verzeichnis den folgenden Befehl aus.

code .

Pyproject.toml aktualisieren

  1. Öffnen Sie die Datei pyproject.toml in Ihrem Editor. Diese Datei enthält die Metadaten für Ihr Projekt.

  2. Überprüfen Sie den Inhalt der Datei. Es sollte mit diesem Beispiel vergleichbar sein. Beachten Sie die Python-Version und die Abhängigkeit; verwenden Sie für mssql-python>=, um eine Mindestversion festzulegen. Wenn Sie eine genaue Version bevorzugen, ändern Sie die >= Vorversionsnummer in ==. Die aufgelösten Versionen der einzelnen Pakete werden dann im uv.lock gespeichert. Die Lockfile stellt sicher, dass Entwickler, die am Projekt arbeiten, konsistente Paketversionen verwenden. Committen Sie beide pyproject.toml und uv.lock und führen Sie einen von der Organisation genehmigten Abhängigkeitsscanner in CI durch. Bearbeiten Sie die uv.lock Datei nicht direkt.

    [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. Aktualisieren Sie den Wert description.

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

  4. Speichern und schließen Sie die Datei.

Aktualisieren von main.py

  1. Öffnen Sie die Datei mit dem Namen main.py. Es sollte mit diesem Beispiel vergleichbar sein.

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

if __name__ == "__main__":
    main()

  2. Ersetzen Sie die gesamten Inhalte von main.py durch den folgenden 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()

Speichern der Verbindungszeichenfolge

  1. Öffnen Sie die .gitignore Datei, und fügen Sie einen Ausschluss für Dateien hinzu .env . Ihre Datei sollte dem folgenden Beispiel ähneln.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

  2. Speichern Sie und schließen Sie .gitignore.

  3. Erstellen Sie im aktuellen Verzeichnis eine neue Datei mit dem Namen .env.

  4. Fügen Sie in der .env Datei einen Eintrag für die Verbindungszeichenfolge mit dem Namen SQL_CONNECTION_STRINGhinzu. Ersetzen Sie die Platzhalter durch Ihre tatsächlichen Werte.

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

    Important

    Halten Sie .env lokal und aus der Versionsverwaltung heraus. Für CI- und bereitgestellte Umgebungen injizieren Sie die Verbindungszeichenfolge oder deren Komponentengeheimnisse aus Ihrem Plattform-Geheimnisspeicher, anstatt .env zwischen Rechnern zu kopieren.

    Die Verbindungszeichenfolge hängt von der Art der SQL-Datenbank ab, mit der du dich verbindest. Für die Azure SQL-Datenbank oder SQL-Datenbank in Fabric verwenden Sie die ODBC-Verbindungszeichenfolge aus dem Tab „Verbindungszeichenfolgen“ und passen Sie den Authentifizierungstyp für Ihr Szenario an. Für weitere Informationen zur Verbindungszeichenfolge-Syntax siehe Connection strings for mssql-python.

    Tip

    Unter macOS funktionieren sowohl ActiveDirectoryInteractive als auch ActiveDirectoryDefault für die Microsoft Entra-Authentifizierung. ActiveDirectoryInteractive fordert Sie auf, sich jedes Mal anzumelden, wenn Sie das Skript ausführen. Um wiederholte Anmeldeaufforderungen zu vermeiden, melden Sie sich einmal mit dem Azure CLIaz login-Befehl an und verwenden Sie ActiveDirectoryDefault. ActiveDirectoryDefault Verwendet die zwischengespeicherten Zugangsdaten wieder.

Verwenden Sie "uv run", um das Skript auszuführen.

Führen Sie im Terminalfenster vor oder in einem neuen Terminalfenster, das im selben Verzeichnis geöffnet ist, den folgenden Befehl aus.

uv run main.py

Das Skript demonstriert drei Arrow-Abrufmethoden:

  • cursor.arrow() gibt eine vollständige Variante pyarrow.Table mit allen Zeilen zurück. Nutze es für kleine bis mittlere Ergebnismengen, bei denen du den gesamten Datensatz im Speicher brauchst.

  • cursor.arrow_batch() gibt jeweils ein pyarrow.RecordBatch zurück. Nutze es für große Ergebnismengen, bei denen du Daten schrittweise verarbeiten möchtest, ohne alles in den Speicher zu laden.

  • cursor.arrow_reader() gibt einen Streaming-Reader zurück, der bei der Iteration pyarrow.RecordBatch-Objekte liefert. Am besten für Pipeline-ähnliche Verarbeitung oder das direkte Weitergeben an Bibliotheken, die einen Leser akzeptieren.

Das Skript speichert außerdem die Produktdaten in einer Parquet-Datei und liest sie wieder ein, um den Roundtrip zu verifizieren.

Funktionsweise des Codes

  1. Verbindung: Das Skript lädt die Verbindungszeichenfolge aus einer .env Datei und erstellt eine Verbindung mit mssql_python.connect().

  2. Vollständiger Tabellenabruf: cursor.arrow() führt die Abfrage aus und gibt die gesamte Ergebnismenge als pyarrow.Table zurück. Der Treiber konvertiert Daten in seiner C++-Schicht über die Arrow C Data Interface und umgeht damit die Erstellung von Python-Objekten für eine bessere Leistung.

  3. Batchstreaming: cursor.arrow_batch() gibt pro Aufruf eine pyarrow.RecordBatch zurück. Die Schleife sammelt Chargen, bis keine Zeilen mehr übrig sind, und kombiniert sie dann zu einer einzigen Tabelle. Verwenden Sie diesen Ansatz für große Datensätze oder wenn Sie jede Charge unabhängig verarbeiten möchten.

  4. Streamingreader: cursor.arrow_reader() Gibt einen Reader zurück, der beim Iterieren pyarrow.RecordBatch-Objekte liefert. Im Vergleich zur arrow_batch()-Schleife verwaltet der Reader seinen eigenen Iterationszustand, sodass du nicht selbst prüfen musst, ob die Ergebnismenge erschöpft ist.

    Diese Funktion liest den Stream vollständig und schließt dann den Cursor, sodass der Reader nicht geschlossen werden muss. Schließe den Leseleser, wenn du frühzeitig aufhörst zu lesen, und benutze den Cursor weiter. Weitere Informationen finden Sie unter Apache Arrow Integration.

  5. Parquet I/O: pyarrow.parquet.write_table() Speichert die Arrow-Tabelle in einer komprimierten Parquet-Datei. Dieses Format bewahrt Spaltentypen und unterstützt effiziente Teillesungen.

Verwandte Inhalte

  • Zuletzt aktualisiert um