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

apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2

dnf install -y libtool-ltdl krb5-libs

zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2

zypper install -y libltdl7

brew install openssl

Tworzenie bazy danych SQL

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

Tworzenie bazy danych SQL w ciągu kilku minut przy użyciu witryny Azure Portal

Tworzenie projektu i uruchamianie kodu

Tworzenie nowego projektu
Dodawanie zależności
Uruchamianie programu Visual Studio Code
Aktualizowanie pliku pyproject.toml
Aktualizowanie main.py
Zapisywanie parametrów połączenia
Użyj narzędzia uv run, aby wykonać skrypt

Utwórz nowy projekt

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

Plik pyproject.toml zawiera metadane twojego projektu. Otwórz plik w ulubionym edytorze.
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.5.0",
    "pyarrow>=19.0.0",
    "python-dotenv>=1.1.1",
    "rich>=14.1.0",
]
```

Zaktualizuj opis, aby był bardziej opisowy.

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

Zapisz i zamknij plik.

Aktualizowanie main.py

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

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

Zapisz łańcuch połączeniowy

.gitignore Otwórz plik i dodaj wykluczenie dla .env plików. Plik powinien być podobny do tego przykładu. Pamiętaj, aby zapisać i zamknąć go po zakończeniu.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

W bieżącym katalogu utwórz nowy plik o nazwie .env.
W pliku .env dodaj wpis dla łańcucha połączenia o nazwie SQL_CONNECTION_STRING. Zastąp przykład wartością rzeczywistych parametrów połączenia.
```
SQL_CONNECTION_STRING="Server=<server_name>;Database=<database_name>;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.

Wskazówka

Używany przez Ciebie parametry połączenia zależy w dużej mierze od typu bazy danych SQL, do której się łączysz. Jeśli nawiązujesz połączenie z usługą Azure SQL Database lub bazą danych SQL w sieci szkieletowej, użyj parametrów połączenia ODBC z karty parametry połączenia. W zależności od scenariusza może być konieczne dostosowanie typu uwierzytelniania. Aby uzyskać więcej informacji na temat parametrów połączenia i ich składni, zobacz dokumentację składni parametrów połączenia.

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

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ę monitów logowania, zaloguj się raz za pomocą Azure CLI, uruchamiając az login, a następnie użyj ActiveDirectoryDefault, które ponownie wykorzystuje poświadczenia zapisane w pamięci podręcznej.

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. Najlepsze dla małych i średnich zestawów wyników, gdzie potrzebny jest pełny zestaw danych w pamięci.
- cursor.arrow_batch() zwraca po jednym pyarrow.RecordBatch naraz. Najlepsze do dużych zestawów wyników, gdzie chcesz przetwarzać dane stopniowo, bez ładowania wszystkiego do pamięci.
- cursor.arrow_reader() zwraca pyarrow.RecordBatchReader do przesyłania strumieniowego. Najlepsze 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

Połączenie: Skrypt ładuje parametry połączenia z .env pliku i tworzy połączenie za pomocą mssql_python.connect().
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.
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.
RecordBatchReader: cursor.arrow_reader() zwraca pyarrow.RecordBatchReader, standardowy interfejs Arrow, który wiele bibliotek akceptuje bezpośrednio. Wywołanie reader.read_all() powoduje zapisanie całego strumienia do tabeli.
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.

Następne kroki

Wykorzystaj te artykuły, aby dalej budować:

Integracja Arrow dla zaawansowanych wzorców Arrow, w tym przetwarzania wsadowego, zarządzania pamięcią i interoperacji bibliotek.
Integracja z pandas do ładowania wyników zapytań bezpośrednio do DataFrames.
Integracja z Polars do tworzenia ramek danych Polars na podstawie zapytań natywnych dla Arrow.

Integracja z Arrow

Opinia

Czy ta strona była pomocna?

Last updated on 2026-07-22