Gyorsindítás: Apache Arrow az mssql-python driverrel a Python számára

Ebben a rövid útmutatóban azt mutatjuk be, hogyan használhatja a mssql-python illesztőprogram beépített Arrow-lekérési metódusait az SQL Server-adatok Apache Arrow oszlopos tábláiként való lekéréséhez. Az Arrow oszlopos memóriaformátuma lehetővé teszi a nagy teljesítményű analitikát, a másolásmentes együttműködést a pandas, a Polars és a DuckDB között, valamint a Parquet-fájlok hatékony be- és kimenetét Python-objektumok soronkénti létrehozása nélkül.

Az mssql-python illesztőprogram nem igényel külső függőségeket a Windows rendszerű gépeken. Az illesztőprogram egyetlen pip telepítéssel mindent telepít, amire szüksége van, így az illesztőprogram legújabb verzióját használhatja az új szkriptekhez anélkül, hogy tönkretenné azokat a többi szkripteket, amelyek frissítésére és tesztelésére nincs ideje.

az mssql-python dokumentációja | mssql-python forráskód | Csomag (PyPI) | Uv

Prerequisites

Egyszeri operációs rendszerspecifikus előfeltételek telepítése. A Windows felhasználók ezt a lépést kihagyhatják. A platform teljes részleteiért lásd: Install mssql-python.

apk add libtool krb5-libs krb5-dev

SQL-adatbázis létrehozása

Létrehozni vagy csatlakozni SQL adatbázishoz az alábbi platformok egyikén:

A projekt létrehozása és a kód futtatása

  1. Új projekt létrehozása
  2. Függőségek hozzáadása
  3. A Visual Studio Code indítása
  4. Pyproject.toml frissítése
  5. Main.py frissítése
  6. A kapcsolati sztring mentése
  7. A szkript végrehajtása uv-futtatás használatával

Új projekt létrehozása

  1. Nyisson meg egy parancssort a fejlesztői címtárban. Ha nincs ilyened, hozz létre egy új könyvtárat, például python vagy scripts. Kerüld a mappákat a OneDrive-on, mert a szinkronizáció zavarhatja a virtuális környezet kezelését.

  2. Új projektet hozz létre .uv

    uv init arrow-qs
cd arrow-qs

Függőségek hozzáadása

Ugyanabban a könyvtárban telepítsd a mssql-python, python-dotenv, pyarrow, és rich a csomagokat.

uv add mssql-python python-dotenv pyarrow rich

Indítsa el a Visual Studio Code-ot

Ugyanabban a könyvtárban futtassa a következő parancsot.

code .

Pyproject.toml frissítése

  1. A pyproject.toml fájl tartalmazza a projekted metaadatait. Nyissa meg a fájlt a kedvenc szerkesztőjében.

  2. Tekintse át a fájl tartalmát. Ennek hasonlónak kell lennie ehhez a példához. Jegyezze fel a Python-verziót és a függőséget; a >= használatával határozhat meg minimális verziót a(z) mssql-python számára. Ha pontos verziót szeretne, módosítsa a >= verziószám előtti értéket ==. Az egyes csomagok feloldott verzióit ezután az uv.lock tárolja. A lockfile biztosítja, hogy a projekten dolgozó fejlesztők következetes csomagverziókat használjanak. Kommitáld mind a pyproject.toml-t, mind a uv.lock-et, és futtass egy, a szervezet által jóváhagyott függőségi szkennert a CI-ben. Ne szerkessze közvetlenül a uv.lock fájlt.

    [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. Módosítsa a leírást részletesebbre.

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

  4. Mentse és zárja be a fájlt.

Main.py frissítése

  1. Nyissa meg a nevű main.pyfájlt. Ennek hasonlónak kell lennie ehhez a példához.

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

if __name__ == "__main__":
    main()

  2. Cserélje le a(z) main.py teljes tartalmát a következő kóddal.

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

A kapcsolati sztring mentése

  1. Nyissa meg a .gitignore fájlt, és adjon hozzá kizárást a fájlokhoz .env . A fájlnak hasonlónak kell lennie ehhez a példához. Mindenképpen mentse és zárja be, ha elkészült.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

  2. Az aktuális könyvtárban hozzon létre egy új fájlt..env

  3. A .env fájlban adjon hozzá egy bejegyzést a kapcsolati karakterlánchoz, a következő névvel: SQL_CONNECTION_STRING. Cserélje le az itt látható példát a tényleges kapcsolati karakterláncértékre.

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

    Important

    .env maradjon lokálisan, és ne kerüljön verziókezelés alá. CI- és telepített környezetek esetén a gépek közötti .env másolás helyett add meg a kapcsolódási karakterláncot vagy az összetevőinek titkos adatait a platform titoktárolójából.

    Jótanács

    A használt kapcsolati karakterlánc nagyrészt attól függ, milyen típusú SQL adatbázishoz csatlakozol. Ha Azure SQL Database-hez vagy SQL-adatbázishoz csatlakozik a Fabricben, használja az ODBC kapcsolati sztringet a kapcsolati sztringek lapról. Előfordulhat, hogy a forgatókönyvtől függően módosítania kell a hitelesítési típust. A kapcsolati sztringekről és azok szintaxisáról további információt a kapcsolati sztring szintaxisának hivatkozásában talál.

Használja az "uv run" parancsot a szkript futtatásához

Jótanács

macOS rendszeren mind a ActiveDirectoryInteractive, mind a ActiveDirectoryDefault használható a Microsoft Entra-hitelesítéshez. ActiveDirectoryInteractive minden alkalommal, amikor futtatja a szkriptet, arra kéri, hogy jelentkezzen be. Az ismétlődő bejelentkezési promptok elkerülése érdekében egyszer jelentkezz be az Azure CLI-n keresztül a az loginfuttatással , majd használd ActiveDirectoryDefault, amely újrahasználja a gyorsítótározott hitelesítő adatokat.

  • A korábban megnyitott terminálablakban vagy egy új terminálablakban futtassa a következő parancsot.

    uv run main.py

    A szkript három Arrow fetch módszert mutat be:

    • cursor.arrow() egy teljes pyarrow.Table értéket ad vissza az összes sorral. A legjobb kis vagy közepes eredményhalmazokhoz, ahol a teljes adatkészletre van szükség a memóriában.

    • cursor.arrow_batch() egyszerre egy pyarrow.RecordBatch elemet ad vissza. A legjobb nagy eredményhalmazokhoz, ahol fokozatosan akarod feldolgozni az adatokat anélkül, hogy mindent betöltenünk a memóriába.

    • cursor.arrow_reader() egy pyarrow.RecordBatchReader objektumot ad vissza a streameléshez. A pipeline stílusú feldolgozásra vagy közvetlenül olvasót fogadó könyvtárakba való továbbításra alkalmas.

    A szkript a termékadatokat egy Parquet-fájlba is menti, majd visszaolvassa őket, hogy ellenőrizze az oda-vissza átalakítás helyességét.

A kód működése

  1. Kapcsolat: A szkript betölti a kapcsolati karakterláncot egy .env fájlból, és kapcsolatot hoz létre a(z) mssql_python.connect() használatával.

  2. Teljes tábla beolvasása: cursor.arrow() lefuttatja a lekérdezést, és a teljes eredményhalmazt pyarrow.Table formájában adja vissza. Az illezőprogram a C++ rétegben az Arrow C adatfelületen konvertálja az adatokat, megkerülve a Python objektumalkotást a jobb teljesítmény érdekében.

  3. Kötegelt adatfolyam: cursor.arrow_batch() hívásonként egy pyarrow.RecordBatch-t ad vissza. A hurok összegyűjti a sorozatokat, amíg már nem marad több sor, majd egyesíti őket egyetlen táblázatba. Használd ezt a megközelítést nagy adathalmazokhoz vagy amikor minden kötetet önállóan szeretnél feldolgozni.

  4. RecordBatchReader: cursor.arrow_reader() egy pyarrow.RecordBatchReader, egy szabványos Arrow interfészt ad vissza, amelyet sok könyvtár közvetlenül elfogad. A(z) reader.read_all() meghívása a teljes adatfolyamot táblába olvassa.

  5. Parquet I/O: pyarrow.parquet.write_table() elmenti az Arrow táblát egy tömörített Parquet fájlba. Ez a formátum megőrzi az oszloptípusokat, és hatékony részleges olvasásokat támogat.

Következő lépések

Használd ezeket a cikkeket, hogy tovább építsd:

  • Arrow-integráció speciális Arrow-mintákhoz, beleértve a kötegelt feldolgozást, a memóriakezelést és a könyvtárak közötti interoperabilitást.
  • pandas-integráció a lekérdezési eredmények közvetlenül DataFrame-ekbe való betöltéséhez.
  • Polars-integráció Polars DataFrame-ek létrehozásához Arrow-natív lekérdezésekből.

Nyíl integráció

  • Last updated on