Hızlı Başlangıç: Apache Arrow ve Python için mssql-python sürücüsü

Bu hızlı başlangıçta, mssql-python sürücünün yerleşik Arrow fetch yöntemlerini kullanarak SQL Server verilerini sütunlu Apache Arrow tabloları olarak alın. Arrow'un sütunlu bellek formatı, yüksek performanslı analitik, pandalar, Polars ve DuckDB ile sıfır kopya etkileşimi ve satır satır Python nesne oluşturmadan verimli Parquet dosya giriş/girişi sağlar.

Sürücü, mssql-python Windows makinelerinde dış bağımlılık gerektirmez. Sürücü, ihtiyaç duyduğu her şeyi tek bir pip kurulumuyla yükler; böylece, yükseltip test etmeye vakit bulamadığın diğer betikleri bozmadan, yeni betikler için sürücünün en son sürümünü kullanabilirsin.

mssql-python belgeleri | mssql-python kaynak kodu | Paket (PyPI) | Uv

Prerequisites

Tek seferlik işletim sistemine özgü önkoşulları yükleyin. Windows kullanıcıları bu adımı atlayabilir. Platformun tam detayları için mssql-python'u Install sayfasına bakınız.

apk add libtool krb5-libs krb5-dev

SQL veritabanı oluşturma

Aşağıdaki platformlardan birinde bir SQL veritabanı oluşturun veya bağlanın:

Projeyi oluşturma ve kodu çalıştırma

  1. Yeni proje oluşturma
  2. Bağımlılık ekleme
  3. Visual Studio Code'ı başlatma
  4. pyproject.toml güncelleştirme
  5. Main.py'yi Güncelle
  6. Bağlantı dizesini kaydetme
  7. Betik dosyasını çalıştırmak için uv run kullanın

Yeni proje oluşturma

  1. Geliştirme dizininizde bir komut istemi açın. Eğer bir dizin yoksa, yeni bir dizin oluşturun, örneğin python veya scripts. OneDrive'ınızdaki klasörlerden kaçının, çünkü senkronizasyon sanal ortamınızı yönetmenize engel olabilir.

  2. uv kullanarak yeni bir proje oluşturun.

    uv init arrow-qs
cd arrow-qs

Bağımlılık ekleme

Aynı dizinde , mssql-python, python-dotenv, ve pyarrow paketleri yükleyinrich.

uv add mssql-python python-dotenv pyarrow rich

Visual Studio Code'ı başlatma

Aynı dizinde aşağıdaki komutu çalıştırın.

code .

pyproject.toml güncelle

  1. pyproject.toml dosyası projenizin meta verilerini içerir. Dosyayı sık kullandığınız düzenleyicide açın.

  2. Dosyanın içeriğini gözden geçirin. Bu örneğe benzer olmalıdır. mssql-python için Python sürümüne ve bağımlılığına dikkat edin; en düşük sürümü tanımlamak için >= kullanın. Tam sürümü tercih ediyorsanız, sürüm numarasının önündeki >='yi == ile değiştirin. Her paketin çözümlenen sürümleri daha sonra uv.lock dosyasında depolanır. Kilit dosyası, projede çalışan geliştiricilerin tutarlı paket sürümlerini kullanmasını sağlar. pyproject.toml ve uv.lock öğelerinin ikisini de commit edin ve CI'de kuruluş tarafından onaylanmış bir bağımlılık tarayıcısı çalıştırın. uv.lock dosyasını doğrudan düzenlemeyin.

    [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. Açıklamayı daha açıklayıcı olacak şekilde güncelleştirin.

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

  4. Dosyayı kaydedin ve kapatın.

Güncelle main.py

  1. adlı main.pydosyayı açın. Bu örneğe benzer olmalıdır.

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

if __name__ == "__main__":
    main()

  2. içindekilerin main.py tamamını aşağıdaki kodla değiştirin.

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

Bağlantı dizesini kaydetme

  1. .gitignore dosyasını açın ve .env dosyaları için bir dışlama ekleyin. Dosyanız bu örneğe benzer olmalıdır. İşiniz bittiğinde kaydetmeyi ve kapatmayı unutmayın.

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

# Virtual environments
.venv

# Connection strings and secrets
.env

# Generated data files
*.parquet

  2. Geçerli dizinde adlı .envyeni bir dosya oluşturun.

  3. dosyasının içine .env bağlantı dizeniz için adlı SQL_CONNECTION_STRINGbir girdi ekleyin. Buradaki örneği gerçek bağlantı dizesi değerinizle değiştirin.

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

    Important

    .env öğesini yerel tutun ve sürüm denetiminden uzak tutun. CI ve dağıtıma alınmış ortamlar için, .env öğesini makineler arasında kopyalamak yerine bağlantı dizesini veya onu oluşturan sırları platformunuzun sır deposundan enjekte edin.

    Tip

    Kullandığınız bağlantı dizesi, büyük ölçüde bağlandığınız SQL veritabanı türüne bağlıdır. Eğer bir Azure SQL Veritabanına veya Fabric'teki SQL veritabanına bağlanıyorsanız, bağlantı dizeleri sekmesindeki ODBC bağlantı dizesini kullanın. Senaryonuza bağlı olarak kimlik doğrulama türünü ayarlamanız gerekebilir. Bağlantı dizeleri ve söz dizimi hakkında daha fazla bilgi için bkz. bağlantı dizesi söz dizimi başvurusu.

Betiği yürütmek için uv çalıştır komutunu kullan

Tip

macOS'ta hem ActiveDirectoryInteractive hem de ActiveDirectoryDefault Microsoft Entra kimlik doğrulaması için çalışır. ActiveDirectoryInteractive betiği her çalıştırdığınızda oturum açmanızı ister. Tekrar tekrar giriş çağrılarını önlemek için, Azure CLI üzerinden bir kez oturum açarak az login, çalıştırın, ardından önbelleklenmiş kimlik bilgilerini yeniden kullanan 'i kullanınActiveDirectoryDefault.

  • Önceki terminal penceresinde veya aynı dizine açılan yeni bir terminal penceresinde aşağıdaki komutu çalıştırın.

    uv run main.py

    Script, üç Arrow getirme yöntemini gösterir:

    • cursor.arrow() Tüm satırlarla birlikte tam bir pyarrow.Table döndürür. Tüm veri setinin hafızada olmasına ihtiyaç duyduğunuz küçük ve orta seviye sonuç setleri için en iyisi.

    • cursor.arrow_batch(), her seferinde bir pyarrow.RecordBatch döndürür. Her şeyi belleğe yüklemeden veriyi kademeli olarak işlemek istediğiniz büyük sonuç setleri için en iyisi.

    • cursor.arrow_reader(), akış için bir pyarrow.RecordBatchReader döndürür. Pipeline tarzı işleme veya okuyucu kabul eden kütüphanelere doğrudan aktarma için en iyisidir.

    Script ayrıca ürün verilerini bir Parquet dosyasına kaydeder ve dönüşü doğrulamak için geri okur.

Kod nasıl çalışır?

  1. Bağlantı: Betik, bağlantı dizesini bir .env dosyasından yükler ve mssql_python.connect() kullanarak bir bağlantı oluşturur.

  2. Tam tablo getirme: cursor.arrow() sorguyu çalıştırır ve tüm sonuç kümesini bir pyarrow.Tableolarak döndürür. Sürücü, C++ katmanındaki verileri Arrow C Veri Arayüzü kullanarak dönüştürür ve performansı artırmak için Python nesne oluşturmayı atlar.

  3. Toplu akış: cursor.arrow_batch() her çağrıda bir pyarrow.RecordBatch döndürür. Döngü, artık satır kalmayana kadar grupları toplar, ardından bunları tek bir tabloda birleştirir. Bu yaklaşımı büyük veri kümeleri için veya her partiyi bağımsız olarak işlemek istediğinizde kullanın.

  4. RecordBatchReader: cursor.arrow_reader(), birçok kitaplık tarafından doğrudan kabul edilen standart bir Arrow arayüzü olan bir pyarrow.RecordBatchReader döndürür. reader.read_all() çağrısı, tüm akışı tüketerek bir tabloya dönüştürür.

  5. Parquet I/O: pyarrow.parquet.write_table() Ok tablosunu sıkıştırılmış bir Parquet dosyasına kaydeder. Bu format, sütun türlerini korur ve verimli kısmi okumaları destekler.

Sonraki Adımlar

İnşaya devam etmek için bu makaleleri kullanın:

  • Arrow entegrasyonu; toplu işleme, bellek yönetimi ve kütüphaneler arası birlikte çalışabilirlik dahil olmak üzere gelişmiş Arrow desenleri için.
  • sorgu sonuçlarını doğrudan DataFrame’lere yüklemek için pandas entegrasyonu.
  • Polars entegrasyonu: Arrow yerel sorgularından Polars veri çerçeveleri oluşturmak için.

Arrow entegrasyonu

  • Last updated on