Gunakan mssql-python dengan DuckDB

DuckDB adalah mesin analitik SQL dalam proses yang dapat mengkueri tabel Apache Arrow secara langsung tanpa menyalin data. Menggabungkan DuckDB dengan driver mssql-python memungkinkan Anda:

  • Jalankan kueri SQL analitik langsung pada kumpulan hasil Microsoft SQL tanpa memuat data ke pandas atau Polars.
  • Lakukan kueri pada tabel Arrow di memori tanpa overhead penyalinan.
  • Gabungkan data Microsoft SQL dengan file lokal (CSV, Parquet, JSON) dalam satu kueri DuckDB.
  • Ekspor data Microsoft SQL ke Parquet, CSV, atau format lain melalui DuckDB.

Prasyarat

  • Python 3.10 atau yang lebih baru.
  • Paket mssql-python, duckdb, dan pyarrow. Instal semuanya dengan pip install mssql-python duckdb pyarrow.
  • Instal prasyarat khusus sistem operasi satu kali. Pengguna Windows dapat melewati langkah ini. Untuk detail platform lengkap, lihat Menginstal mssql-python.
    apk add libtool krb5-libs krb5-dev

Membuat database SQL

Buat atau sambungkan ke database SQL di salah satu platform berikut:

Contoh dalam artikel ini menggunakan kueri pada database sampel AdventureWorks ini. Jika Anda belum memilikinya, lihat contoh database AdventureWorks.

Pasang dependensi

pip install mssql-python duckdb pyarrow

Mengkueri data Microsoft SQL dengan DuckDB

Alur kerja dasarnya adalah: jalankan kueri dengan mssql-python, ambil hasilnya sebagai tabel Arrow, lalu kueri tabel Arrow tersebut dengan DuckDB SQL.

Pola dasar

Mulailah dengan membuat sambungan dan mengambil data sebagai tabel 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 mereferensikan products tabel Arrow dengan nama variabel Python-nya. Tidak ada data yang disalin ke penyimpanan DuckDB.

Gabungkan dan saring

Gunakan SQL DuckDB untuk mengelompokkan dan menggabungkan data 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())

Menggabungkan beberapa hasil Microsoft SQL

Ambil beberapa tabel dari Microsoft SQL dan gabungkan mereka di DuckDB tanpa menulis kueri lintas server.

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

Menggabungkan data SQL Microsoft dengan file lokal

DuckDB dapat membaca file CSV, Parquet, dan JSON secara asli. Gabungkan data SQL Server dengan file lokal dalam satu kueri.

Gabungkan dengan file CSV

Muat file CSV dan gabungkan dengan data dari 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)

Gabungkan dengan file Parquet

Muat file Parquet dan gabungkan dengan data dari 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)

Mengekspor data Microsoft SQL

Gunakan pernyataan DuckDB COPY untuk mengekspor data Microsoft SQL ke berbagai format file.

Ekspor ke Parket

Ekspor data ke format Apache Parquet.

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

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

Ekspor ke CSV

Mengekspor data ke file nilai yang dipisahkan koma:

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

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

Ekspor Parket yang dipartisi

Mengekspor data ke file Parquet yang dipartisi untuk analitik terdistribusi:

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

Alirkan set hasil berukuran besar

Untuk himpunan data besar, gunakan arrow_reader() untuk memproses data dalam batch streaming tanpa memuat semua baris ke memori sekaligus:

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

Akumulasikan hasil streaming

Untuk menggabungkan di semua batch, daftarkan setiap batch dalam koneksi DuckDB persisten dan akumulasi hasil secara bertahap.

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

Tips kinerja

Biarkan Microsoft SQL menangani pekerjaan berat

Microsoft SQL lebih cepat untuk pemfilteran, gabungan, dan agregasi daripada menarik semua data mentah melalui kabel. Gunakan DuckDB untuk analisis sekunder pada kumpulan hasil yang sudah diambil, bukan sebagai pengganti pengoptimalan kueri 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")

Gunakan Arrow untuk semua operasi pembacaan

Transfer berbasis panah menghindari pembuatan objek Python perantara, yang mengurangi penggunaan memori dan meningkatkan throughput. Utamakan cursor.arrow() daripada konversi manual per baris saat meneruskan data ke DuckDB.

Menggunakan streaming untuk himpunan data besar

Untuk kumpulan hasil yang lebih besar dari memori yang tersedia, gunakan arrow_reader() dengan batch_size parameter untuk memproses data secara bertahap.

Konten terkait

  • Last updated on