Użyj mssql-python z DuckDB

DuckDB to działający w procesie analityczny silnik SQL, który może bezpośrednio wykonywać zapytania na tabelach Apache Arrow bez kopiowania danych. Połączenie DuckDB ze sterownikiem mssql-python pozwala Ci:

  • Uruchamiaj analityczne zapytania SQL na wynikach Microsoft SQL Server bez ładowania danych do pandas lub Polars.
  • Wykonuj zapytania na tabelach Arrow w pamięci bez narzutu związanego z kopiowaniem danych.
  • Połącz dane Microsoft SQL z plikami lokalnymi (CSV, Parquet, JSON) w jednym zapytaniu DuckDB.
  • Eksportuj dane Microsoft SQL do Parquet, CSV lub innych formatów przez DuckDB.

Wymagania wstępne

  • Python 3.10 lub nowszy.
  • Pakiety mssql-python, duckdb i pyarrow. Zainstaluj wszystko za pomocą pip install mssql-python duckdb pyarrow.
  • 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

Tworzenie bazy danych SQL

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

Przykłady w tym artykule wykonują zapytania do przykładowej bazy danych AdventureWorks. Jeśli jeszcze go nie masz, zobacz przykładowe bazy danych AdventureWorks.

Instalowanie zależności

pip install mssql-python duckdb pyarrow

Zapytanie do danych Microsoft SQL za pomocą DuckDB

Podstawowy workflow to: wykonanie zapytania w mssql-python, pobranie wyników jako tabeli Arrow, a następnie zapytanie tej tabeli Arrow za pomocą DuckDB SQL.

Podstawowy wzór

Zacznij od nawiązania połączenia i pobrania danych jako tabeli 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 odwołuje się do tabeli Arrow products za pomocą nazwy jej zmiennej w Pythonie. Żadne dane nie są kopiowane do pamięci DuckDB.

Grupowanie i filtrowanie

Użyj SQL DuckDB do grupowania i agregowania danych 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())

Dołącz do wielu wyników Microsoft SQL

Pobierz wiele tabel z Microsoft SQL i połącz je w DuckDB bez pisania zapytania między serwerami.

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

Połącz dane Microsoft SQL z plikami lokalnymi

DuckDB potrafi natywnie czytać pliki CSV, Parquet i JSON. Połącz dane SQL Server z lokalnymi plikami w jednym zapytaniu.

Dołącz za pomocą pliku CSV

Załaduj plik CSV i połącz go z danymi 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)

Dołącz za pomocą pliku Parquet

Załaduj plik Parquet i połącz go z danymi 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)

Eksport danych SQL Microsoft

Użyj instrukcji DuckDBCOPY, aby eksportować dane Microsoft SQL do różnych formatów plików.

Eksportuj do formatu Parquet

Eksportuj dane do formatu Apache Parquet.

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

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

Eksportuj do pliku CSV

Eksport danych do pliku z wartościami rozdzielonymi przecinkami:

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

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

Eksport Parquet podzielony

Eksport danych do partycjonowanych plików Parquet na potrzeby analizy rozproszonej:

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

Przesyłaj strumieniowo duże zbiory wyników

W przypadku dużych zbiorów danych użyj arrow_reader() do przetwarzania danych strumieniowo, partiami, bez ładowania wszystkich wierszy do pamięci jednocześnie:

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

Zbieraj wyniki streamingu

Aby agregować dane ze wszystkich partii, rejestruj każdą partię w utrzymywanym połączeniu z DuckDB i narastająco gromadź wyniki.

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

Porady dotyczące wydajności

Pozwól Microsoft SQL wykonać ciężką pracę

Microsoft SQL jest szybszy w filtrowaniu, łączeniu i agregacji niż pobieranie wszystkich surowych danych przez linię internetową. Używaj DuckDB do analizy wtórnej na już pobranych zestawach wyników, a nie jako zamiennika optymalizacji zapytań 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")

Używaj Arrow do wszystkich operacji odczytu

Transfer oparty na strzałkach unika tworzenia pośrednich obiektów Python, co zmniejsza zużycie pamięci i poprawia przepustowość. Preferuj cursor.arrow() zamiast ręcznej konwersji wiersz po wierszu podczas przekazywania danych do DuckDB.

Używaj streamingu dla dużych zbiorów danych

W przypadku zestawów wyników większych niż dostępna pamięć użyj arrow_reader() z parametrem batch_size, aby przetwarzać dane stopniowo.

Treści powiązane

  • Last updated on