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

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

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.

  1. Ein neues Projekt erstellen
  2. Hinzufügen von Abhängigkeiten
  3. Starten von Visual Studio Code
  4. Pyproject.toml aktualisieren
  5. Aktualisieren von main.py
  6. Speichern der Verbindungszeichenfolge
  7. Verwenden Sie 'uv run', um das Skript auszuführen

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. Die Datei pyproject.toml enthält die Metadaten für Ihr Projekt. Öffnen Sie die Datei in Ihrem bevorzugten Editor.

  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. Mach beide pyproject.toml und uv.lock, und führe 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.5.0",
    "pyarrow>=19.0.0",
    "python-dotenv>=1.1.1",
    "rich>=14.1.0",
]

  3. Aktualisieren Sie die Beschreibung so, dass sie aussagekräftiger ist.

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

Speichern der Verbindungszeichenfolge

  1. Öffnen Sie die .gitignore Datei, und fügen Sie einen Ausschluss für Dateien hinzu .env . Ihre Datei sollte mit diesem Beispiel vergleichbar sein. Achten Sie darauf, sie zu speichern und zu schließen, wenn Sie fertig sind.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

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

  3. Fügen Sie in der .env Datei einen Eintrag für die Verbindungszeichenfolge mit dem Namen SQL_CONNECTION_STRINGhinzu. Ersetzen Sie das Beispiel hier durch Ihren tatsächlichen Verbindungszeichenfolgenwert.

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

    Important

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

    Tip

    Der verwendete Verbindungszeichenfolge hängt stark von der Art der SQL-Datenbank ab, mit der du dich verbindest. Wenn Sie eine Verbindung mit einer Azure SQL-Datenbank oder einer SQL-Datenbank in Fabric herstellen, verwenden Sie die ODBC-Verbindungszeichenfolge auf der Registerkarte "Verbindungszeichenfolgen". Möglicherweise müssen Sie den Authentifizierungstyp je nach Szenario anpassen. Weitere Informationen zu Verbindungszeichenfolgen und deren Syntax finden Sie in der Referenz zur Verbindungszeichenfolgensyntax.

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

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 über die Azure CLI an, indem Sie az login ausführen, und verwenden Sie dann ActiveDirectoryDefault, das die zwischengespeicherten Anmeldeinformationen wiederverwendet.

  • 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. Ideal für kleine bis mittlere Ergebnismengen, wenn der vollständige Datensatz im Speicher benötigt wird.

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

    • cursor.arrow_reader() gibt zum Streamen ein pyarrow.RecordBatchReader zurück. Am besten für Pipeline-ähnliche Verarbeitung oder das direkte Weitergeben an Bibliotheken, die einen Reader 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. RecordBatchReader: cursor.arrow_reader() gibt eine pyarrow.RecordBatchReader, eine Standard-Arrow-Schnittstelle zurück, die viele Bibliotheken direkt akzeptieren. Der Aufruf von reader.read_all() liest den gesamten Stream in eine Tabelle ein.

  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.

Nächste Schritte

Nutzen Sie diese Artikel, um weiter aufzubauen:

  • Integration von Arrow für fortgeschrittene Arrow-Anwendungsmuster, einschließlich Batchverarbeitung, Speicherverwaltung und Interoperabilität mit Bibliotheken.
  • pandas-Integration zum Laden von Abfrageergebnissen direkt in DataFrames.
  • Polars-Integration zum Aufbau von Polars DataFrames aus Arrow-nativen Abfragen.

Pfeilintegration

  • Last updated on