使用 mssql-python 搭配 DuckDB

DuckDB 是一個正在進行的 SQL 分析引擎，可以直接查詢 Apache Arrow 資料表，且不複製資料。 結合 DuckDB 與 mssql-python 驅動程式，可以：

• 在不載入 PANDAS 或 Polar 資料的情況下，對 Microsoft SQL 結果集執行分析性 SQL 查詢。
• 查詢記憶體中零複製開銷的 Arrow 表格。
• 在單一 DuckDB 查詢中，將 Microsoft SQL 資料與本地檔案（CSV、Parquet、JSON）結合。
• 透過 DuckDB，將 Microsoft SQL 資料匯出為 Parquet、CSV 或其他格式。

先決條件

• Python 3.10 或更新版本。
• mssql-python、duckdb 和 pyarrow 套件。 使用 pip install mssql-python duckdb pyarrow 安裝全部。
• 安裝一次性作業系統特定先決條件。 Windows 使用者可以跳過此步驟。 完整平台細節請參見 安裝 mssql-python。
  高山

  apk add libtool krb5-libs krb5-dev
  

  Debian/Ubuntu

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

  RHEL

  dnf install -y libtool-ltdl krb5-libs
  

  SUSE

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

  openSUSE

  zypper install -y libltdl7
  

  macOS

  brew install openssl
  

建立 SQL 資料庫

在以下平台建立或連接 SQL 資料庫：

Azure SQL Database

使用 Azure 入口網站在幾分鐘內建立 SQL 資料庫

Fabric 中的 SQL 資料庫

在 Microsoft Fabric 的 SQL 資料庫中載入 AdventureWorks 範例資料

Microsoft SQL 伺服器

從安裝精靈安裝 SQL Server

AdventureWorks 範例資料庫

SQL Server 容器

在 Docker 中建立 SQL Server 容器

用 sqlcmd 建立 SQL Server 容器

建立一個帶有 VS Code MSSQL 擴充功能的 SQL Server 容器

本文中的範例會查詢 AdventureWorks 範例資料庫。 如果你還沒有，請參考 AdventureWorks 的範例資料庫。

安裝依賴項

pip install mssql-python duckdb pyarrow

使用 DuckDB 查詢 Microsoft SQL 資料

基本工作流程是：先用 mssql-python 執行查詢，取得結果作為 Arrow 資料表，然後用 DuckDB SQL 查詢該 Arrow 資料表。

基本模式

首先建立連線並以 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 以 products Python 變數名稱來參考 Arrow 表格。 不會將資料複製到 DuckDB 的儲存空間。

彙總與篩選

使用 DuckDB 的 SQL 來分組和彙整 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())

加入多個 Microsoft SQL 結果

從 Microsoft SQL 抓取多個資料表，並在 DuckDB 中加入，無需寫跨伺服器查詢。

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

將 Microsoft SQL 資料與本地檔案連接

DuckDB 原生可讀取 CSV、Parquet 和 JSON 檔案。 將 SQL Server 資料與本地檔案合併於單一查詢中。

用 CSV 檔案加入

載入一個 CSV 檔案，並用 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)

與 Parquet 檔案聯結

載入 Parquet 檔案，並將其與來自 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)

匯出 Microsoft SQL 資料

使用 DuckDB 的COPY語句將 Microsoft SQL 資料匯出成各種檔案格式。

出口至Parquet

將資料匯出為 Apache Parquet 格式。

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

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

匯出成 CSV

將資料匯出為逗號分隔值檔案：

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

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

匯出分區的 Parquet 檔案

將資料匯出為供分散式分析使用的分割 Parquet 檔案：

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

串流大型結果集

對於大型資料集，請使用 arrow_reader() 串流批次處理資料，避免一次將所有資料列載入記憶體：

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

累積串流結果

要在所有批次間彙整，請將每個批次註冊在持續存在的 DuckDB 連線中，並逐步累積結果。

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

效能祕訣

讓 Microsoft SQL 來處理繁重的工作

Microsoft SQL 在篩選、聯結和彙總方面，比透過網路傳輸拉取所有原始資料更快。 使用 DuckDB 進行已擷取結果集的次級分析，而非取代 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")

所有讀取操作都使用 Arrow

基於箭頭的傳輸避免建立中間的 Python 物件，減少記憶體使用並提升吞吐量。 將資料傳遞給 DuckDB 時，優先使用 cursor.arrow()，而非手動逐列轉換。

使用串流處理大型資料集

對於超出可用記憶體的結果集，使用 arrow_reader() 參數 batch_size 進行資料增量處理。

相關內容

• Apache Arrow 整合
• Pandas整合
• 極座標積分
• 批量複製作業
• 性能調校