Szybki start: Apache Arrow ze sterownikiem mssql-python dla Python

W tym szybkim starcie użyj wbudowanych metod pobierania strzałek sterownika mssql-python do pobierania danych SQL Server jako kolumnowych tabel Apache Arrow. Kolumnowy format pamięci Arrow umożliwia analitykę o wysokiej wydajności, interoperacyjność bez kopiowania danych z pandas, Polars i DuckDB oraz wydajną obsługę wejścia/wyjścia plików Parquet bez konieczności tworzenia obiektów Pythona wiersz po wierszu.

Sterownik mssql-python nie wymaga żadnych zależności zewnętrznych na maszynach z systemem Windows. Sterownik instaluje wszystko, czego potrzebuje, za jedną pip instalacją, więc możesz używać najnowszej wersji sterownika do nowych skryptów bez psucia innych skryptów, których nie masz czasu zaktualizować i przetestować.

Dokumentacja mssql-python Kod źródłowy mssql-python Pakiet (PyPI) uv

Wymagania wstępne

  • Python 3.10 lub nowszy.

  • Jeśli nie masz jeszcze języka Python, zainstaluj środowisko uruchomieniowe języka Python i menedżera pakietów z python.org.

  • Nie chcesz używać własnego środowiska? Postępuj zgodnie z instrukcjami w sekcji Kontenery i programowanie lokalne, aby utworzyć odtwarzalne środowisko devcontainer lub GitHub Codespaces.

  • Visual Studio Code z następującymi rozszerzeniami:

    • Rozszerzenie języka Python dla programu Visual Studio Code

  • Interfejs azure Command-Line (CLI) na potrzeby uwierzytelniania bez hasła w systemach macOS i Linux.

  • Jeśli jeszcze nie masz uvprogramu , postępuj zgodnie z instrukcjami instalacji.

  • Baza danych w programie SQL Server, usłudze Azure SQL Database lub bazie danych SQL w usłudze Fabric z przykładowym AdventureWorks2025 schematem i prawidłowymi parametrami połączenia.

Zainstaluj wymagania wstępne systemu operacyjnego

Zainstaluj jednorazowe wymagania wstępne dotyczące systemu operacyjnego. Użytkownicy Windows mogą pominąć ten krok. Pełne szczegóły dotyczące platformy można znaleźć w artykule Install mssql-python.

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

Tworzenie bazy danych SQL

Stwórz lub połącz się z bazą danych SQL na jednej z następujących platform:

Tworzenie projektu i uruchamianie kodu

Utwórz nowy projekt

  1. Otwórz wiersz polecenia w katalogu deweloperów. Jeśli go nie masz, stwórz nowy katalog, np. python lub scripts. Unikaj folderów na OneDrive, ponieważ synchronizacja może zakłócać zarządzanie środowiskiem wirtualnym.

  2. Stwórz nowy projekt , używając uv.

    uv init arrow-qs
cd arrow-qs

Dodawanie zależności

W tym samym katalogu zainstaluj mssql-python, python-dotenv, pyarrow, oraz rich pakiety.

uv add mssql-python python-dotenv pyarrow rich

Uruchom program Visual Studio Code.

W tym samym katalogu uruchom następujące polecenie.

code .

Aktualizowanie pliku pyproject.toml

  1. pyproject.toml Otwórz plik w edytorze. Ten plik zawiera metadane Twojego projektu.

  2. Przejrzyj zawartość pliku. Powinien być podobny do tego przykładu. Zwróć uwagę na wersję języka Python i zależności; dla mssql-python użyj >=, aby określić minimalną wersję. Jeśli wolisz dokładną wersję, zmień >= wartość przed numerem wersji na ==. Rozwiązane wersje każdego pakietu są następnie przechowywane w uv.lock. Plik blokady zapewnia, że deweloperzy pracujący nad projektem korzystają z jednolitych wersji pakietów. Zatwierdź zarówno pyproject.toml, jak i uv.lock, a następnie uruchom w CI skaner zależności zatwierdzony przez organizację. Nie edytuj uv.lock pliku bezpośrednio.

    [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. Aktualizuj wartość description .

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

  4. Zapisz i zamknij plik.

Aktualizowanie main.py

  1. Otwórz plik o nazwie main.py. Powinien być podobny do tego przykładu.

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

if __name__ == "__main__":
    main()

  2. Zastąp całą zawartość main.py następującym kodem.

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

Zapisz łańcuch połączeniowy

  1. .gitignore Otwórz plik i dodaj wykluczenie dla .env plików. Twój plik powinien być podobny do poniższego przykładu.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

  2. Zapisz i zamknij plik .gitignore.

  3. W bieżącym katalogu utwórz nowy plik o nazwie .env.

  4. W pliku .env dodaj wpis dla łańcucha połączenia o nazwie SQL_CONNECTION_STRING. Zastąp symbole zastępcze rzeczywistymi wartościami.

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

    Important

    Trzymaj .env się lokalnie i z dala od kontroli źródeł. W środowiskach CI i wdrożeniowych wprowadzaj parametry połączenia lub ich sekrety składowe z magazynu sekretów platformy, zamiast kopiować .env między maszynami.

    Parametr połączenia zależy od typu bazy danych SQL, z którą się łączysz. Dla Azure SQL Database lub bazy danych SQL w usłudze Fabric użyj parametry połączenia ODBC z zakładki parametry połączenias i dostosuj typ uwierzytelniania do swojego scenariusza. Więcej informacji o składni parametry połączenia można znaleźć w artykule Connection strings for mssql-python.

    Wskazówka

    Zarówno ActiveDirectoryInteractive, jak i ActiveDirectoryDefault działają na systemie macOS do uwierzytelniania Microsoft Entra. ActiveDirectoryInteractive monituje o zalogowanie się przy każdym uruchomieniu skryptu. Aby uniknąć powtarzających się promptów logowania, zaloguj się raz za pomocą polecenia Azure CLIaz login i użyj ActiveDirectoryDefault. ActiveDirectoryDefault ponownie wykorzystuje zapisane dane uwierzytelniające.

Użyj narzędzia uv run, aby wykonać skrypt

W wcześniejszym oknie terminalu lub w nowym oknie terminalu, które jest otwarte w tym samym katalogu, wykonaj następujące polecenie.

uv run main.py

Skrypt pokazuje trzy metody pobierania Arrow:

  • cursor.arrow() zwraca kompletne pyarrow.Table ze wszystkimi wierszami. Używaj go do małych i średnich zestawów wyników, gdzie potrzebujesz pełnego zbioru danych w pamięci.

  • cursor.arrow_batch() zwraca po jednym pyarrow.RecordBatch naraz. Używaj go do dużych zbiorów wyników, gdzie chcesz przetwarzać dane stopniowo, nie ładując wszystkiego do pamięci.

  • cursor.arrow_reader() zwraca czytnik strumieniowy, który podczas jego iteracji generuje obiekty pyarrow.RecordBatch . Używaj go do przetwarzania w stylu pipeline lub bezpośredniego przekazywania do bibliotek akceptujących czytnik.

Skrypt zapisuje także dane produktu do pliku Parquet i odczytuje je, aby zweryfikować podróż w obie strony.

Jak działa kod

  1. Połączenie: Skrypt ładuje parametry połączenia z .env pliku i tworzy połączenie za pomocą mssql_python.connect().

  2. Pełne pobranie tabeli: cursor.arrow() wykonuje zapytanie i zwraca cały zbiór wyników jako pyarrow.Table. Sterownik konwertuje dane w warstwie C++ za pomocą interfejsu danych Arrow C, omijając tworzenie obiektów w Python dla poprawy wydajności.

  3. Przesyłanie strumieniowe wsadowe: cursor.arrow_batch() zwraca jeden pyarrow.RecordBatch na każde wywołanie. Pętla zbiera partie danych, aż nie pozostaną żadne wiersze, a następnie łączy je w jedną tabelę. Stosuj to podejście do dużych zbiorów danych lub gdy chcesz przetwarzać każdą partię osobno.

  4. Czytnik strumieniowy: cursor.arrow_reader() zwraca czytnik, który generuje obiekty pyarrow.RecordBatch podczas jego iteracji. W porównaniu z pętlą arrow_batch() czytnik zarządza własnym stanem iteracji, więc nie musisz samodzielnie sprawdzać, czy zbiór wyników został wyczerpany.

    Ta funkcja odczytuje strumień do końca, a następnie zamyka kursor, więc nie musi zamykać czytnika. Zamknij czytnik, jeśli przestaniesz czytać przed końcem i nadal korzystasz z kursora. Więcej informacji można znaleźć w artykule o integracji z Apache Arrow.

  5. Parquet I/O: pyarrow.parquet.write_table() zapisuje tabelę Arrow do skompresowanego pliku Parquet. Ten format zachowuje typy kolumn i wspiera efektywne odczyty częściowe.

Treści powiązane

  • Ostatnia aktualizacja o