Použijte mssql-python s DuckDB

DuckDB je průběžný SQL analytický engine, který dokáže přímo dotazovat tabulky Apache Arrow bez kopírování dat. Kombinace DuckDB s ovladačem mssql-python vám umožní:

  • Spouštět analytické SQL dotazy nad sadami výsledků Microsoft SQL, aniž by bylo nutné načítat data do pandas nebo Polars.
  • Dotazujte tabulky Arrow v paměti s nulovou režií kopírování.
  • Spojte data Microsoft SQL s lokálními soubory (CSV, Parquet, JSON) v jednom dotazu DuckDB.
  • Exportujte data Microsoft SQL do Parquet, CSV nebo jiných formátů přes DuckDB.

Předpoklady

  • Python 3.10 nebo novější.
  • Balíčky mssql-python, duckdb a pyarrow. Vše nainstalujte pomocí pip install mssql-python duckdb pyarrow.
  • Nainstalujte požadavky specifické pro jednorázový operační systém. Uživatelé Windows mohou tento krok přeskočit. Pro úplné podrobnosti o platformě viz Instalace mssql-python.
    apk add libtool krb5-libs krb5-dev

Vytvoření databáze SQL

Vytvořte nebo se připojte k SQL databázi na jedné z následujících platforem:

Příklady v tomto článku se dotazují na vzorovou databázi AdventureWorks . Pokud ji ještě nemáte, podívejte se na ukázkové databáze AdventureWorks.

Nainstalujte závislosti

pip install mssql-python duckdb pyarrow

Dotaz na data Microsoft SQL pomocí DuckDB

Základní pracovní postup je: spravit dotaz pomocí mssql-python, načíst výsledky jako tabulku Arrow a poté tuto tabulku Arrow dotazovat pomocí DuckDB SQL.

Základní vzor

Začněte navázáním spojení a načtením dat jako tabulky Arrow.

import duckdb
import mssql_python

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes"
)
cursor = conn.cursor()

# Fetch Microsoft SQL data as Arrow
cursor.execute("SELECT * FROM Production.Product WHERE ListPrice > 0")
products = cursor.arrow()

# Query the Arrow table with DuckDB
result = duckdb.sql("""
    SELECT Color, COUNT(*) AS ProductCount, AVG(ListPrice) AS AvgPrice
    FROM products
    GROUP BY Color
    ORDER BY ProductCount DESC
""")
print(result.fetchdf())

DuckDB odkazuje na tabulku products Arrow podle názvu její Python proměnné. Žádná data nejsou kopírována do úložiště DuckDB.

Agregovat a filtrovat

Použijte SQL DuckDB pro seskupování a agregaci dat Arrow.

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()

# Top customers by total spend
top_customers = duckdb.sql("""
    SELECT
        CustomerID,
        COUNT(*) AS OrderCount,
        SUM(TotalDue) AS TotalSpent,
        AVG(TotalDue) AS AvgOrderValue
    FROM orders
    GROUP BY CustomerID
    HAVING SUM(TotalDue) > 10000
    ORDER BY TotalSpent DESC
    LIMIT 20
""")
print(top_customers.fetchdf())

Připojte se k více výsledkům Microsoft SQL

Načtěte více tabulek z Microsoft SQL a spojte je do DuckDB bez psaní dotazu napříč servery.

# Fetch two tables
cursor.execute("SELECT * FROM Production.Product")
products = cursor.arrow()

cursor.execute("SELECT * FROM Production.ProductSubcategory")
subcategories = cursor.arrow()

# Join in DuckDB
result = duckdb.sql("""
    SELECT
        s.Name AS Subcategory,
        COUNT(*) AS ProductCount,
        ROUND(AVG(p.ListPrice), 2) AS AvgPrice
    FROM products p
    JOIN subcategories s ON p.ProductSubcategoryID = s.ProductSubcategoryID
    GROUP BY s.Name
    ORDER BY AvgPrice DESC
""")
print(result.fetchdf())

Spojte data Microsoft SQL s lokálními soubory

DuckDB umí číst soubory CSV, Parquet a JSON nativně. Spojte data ze SQL Server s lokálními soubory v jednom dotazu.

Sloučit pomocí souboru CSV

Načtěte soubor CSV a spojte jej s daty z Microsoft SQL.

import csv
from pathlib import Path

cursor.execute("SELECT CustomerID, PersonID FROM Sales.Customer")
customers = cursor.arrow()

csv_path = Path("customer_regions.csv")
with csv_path.open("w", newline="", encoding="utf-8") as file:
    writer = csv.writer(file)
    writer.writerow(["CustomerID", "Region", "Segment"])
    writer.writerows([
        (1, "West", "Premium"),
        (2, "East", "Standard"),
        (3, "Central", "Basic"),
    ])

try:
    result = duckdb.sql("""
        SELECT c.CustomerID, c.PersonID, f.Region, f.Segment
        FROM customers c
        JOIN read_csv_auto('customer_regions.csv') f ON c.CustomerID = f.CustomerID
    """)
    print(result.fetchdf())
finally:
    csv_path.unlink(missing_ok=True)

Spojit se souborem Parquet

Načtěte soubor Parquet a spojte jej s daty z Microsoft SQL.

from pathlib import Path

import pyarrow as pa
import pyarrow.parquet as pq

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product")
products = cursor.arrow()

parquet_path = Path("order_history.parquet")
order_history = pa.table({
    "ProductID": [1, 2, 680],
    "OrderDate": ["2024-06-01", "2024-03-15", "2024-01-10"],
    "Quantity": [10, 5, 3],
})
pq.write_table(order_history, parquet_path)

try:
    result = duckdb.sql("""
        SELECT p.Name, p.ListPrice, h.OrderDate, h.Quantity
        FROM products p
        JOIN read_parquet('order_history.parquet') h ON p.ProductID = h.ProductID
        WHERE h.OrderDate >= '2024-01-01'
    """)
    print(result.fetchdf())
finally:
    parquet_path.unlink(missing_ok=True)

Export dat z Microsoft SQL

Použijte příkaz DuckDB COPY k exportu dat Microsoft SQL do různých formátů souborů.

Exportovat do Parquetu

Exportovat data do formátu Apache Parquet.

cursor.execute("SELECT * FROM Production.Product")
products = cursor.arrow()

duckdb.sql("COPY products TO 'products.parquet' (FORMAT PARQUET)")

Export do CSV

Exportujte data do souboru s hodnotami oddělenými čárkami:

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()

duckdb.sql("COPY orders TO 'orders.csv' (FORMAT CSV, HEADER)")

Export děleného souboru Parquet

Exportovat data do rozdělených souborů Parquet pro distribuovanou analytiku:

import shutil
from pathlib import Path

cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()

output_dir = Path("sales_data")
shutil.rmtree(output_dir, ignore_errors=True)

duckdb.sql("""
    COPY (SELECT *, YEAR(OrderDate) AS OrderYear FROM orders)
    TO 'sales_data'
    (FORMAT PARQUET, PARTITION_BY (OrderYear))
""")

Streamovat rozsáhlé sady výsledků

U rozsáhlých datových sad použijte arrow_reader() k zpracování dat po dávkách v režimu streamování, aniž by se všechny řádky načítaly do paměti najednou:

cursor.execute("SELECT * FROM Production.TransactionHistory")
reader = cursor.arrow_reader(batch_size=50000)

# Process each batch with DuckDB
total_rows = 0
for batch in reader:
    result = duckdb.sql("""
        SELECT ProductID, SUM(ActualCost) AS TotalCost
        FROM batch
        GROUP BY ProductID
    """)
    total_rows += batch.num_rows
    print(f"Processed {total_rows} rows")

Shrnout výsledky streamování

Pro agregaci napříč všemi dávkami registrujte každou dávku do trvalého DuckDB spojení a výsledky akumulujte postupně.

cursor.execute("SELECT * FROM Production.TransactionHistory")
reader = cursor.arrow_reader(batch_size=50000)

duck = duckdb.connect()
duck.execute("CREATE TABLE transactions (ProductID INT, ActualCost DOUBLE, Quantity INT)")

for batch in reader:
    duck.execute("INSERT INTO transactions SELECT ProductID, ActualCost, Quantity FROM batch")

# Query the accumulated data
result = duck.sql("""
    SELECT ProductID, SUM(ActualCost) AS TotalCost, SUM(Quantity) AS TotalQty
    FROM transactions
    GROUP BY ProductID
    ORDER BY TotalCost DESC
    LIMIT 10
""")
print(result.fetchdf())
duck.close()

Tipy týkající se výkonu

Nechte Microsoft SQL zvládnout těžkou práci

Microsoft SQL je rychlejší pro filtrování, spojení a agregace než stahování všech surových dat přes kabel. Používejte DuckDB pro sekundární analýzu již načtených výsledků, nikoli jako náhradu za optimalizaci dotazů v SQL Server.

# Suboptimal: Pull all rows, filter in DuckDB
cursor.execute("SELECT * FROM Sales.SalesOrderHeader")
orders = cursor.arrow()
result = duckdb.sql("SELECT * FROM orders WHERE TotalDue > 1000")

# Better: Filter in Microsoft SQL, analyze in DuckDB
cursor.execute("SELECT * FROM Sales.SalesOrderHeader WHERE TotalDue > 1000")
orders = cursor.arrow()
result = duckdb.sql("SELECT CustomerID, SUM(TotalDue) FROM orders GROUP BY CustomerID")

Používejte šipku pro všechny čtecí operace

Přenos na základě šipek se vyhýbá vytváření mezipředmětů v Python, což snižuje využití paměti a zlepšuje propustnost. Preferujte cursor.arrow() před manuálním převodem řádků po řádku při přenosu dat do DuckDB.

Používejte streamování pro velké datové sady

Pro sady výsledků větších, než je dostupná paměť, použijte arrow_reader() s parametrem batch_size k inkrementálnímu zpracování dat.

Související obsah

  • Last updated on