8.2 DuckDB: helyi analitikus SQL szerver nélkül

A Northstar orders és order_items tábláit az előző alfejezetben megtisztítottuk, és a sémát Parquet-ben is rögzítettük. Most ugyanazokat a rendelésszintű mutatókat számoljuk ki DuckDB-vel, amelyeket a 7. fejezetben SQLite-tal is használtunk: rendelések száma, tételek száma, kosárérték és szállítási díj összege. A megfigyelési egység a rendelés, illetve a rendelési tétel; a cél, hogy Pythonból és közvetlen fájlból azonos eredményt kapjunk, explicit DuckDB-kapcsolattal, majd SQLite kontrollértékekkel ellenőrizzünk. Előzetes becslés: a tiszta adatban 2400 rendelés és 4796 tétel található, ezért a belső join sorszáma 4796.

8.2.2 Fogalmi magyarázat

connect. A DuckDB beágyazott adatbázis: nincs külön szerverfolyamat, a motor a hívó Python-folyamatban fut. A duckdb.connect(":memory:") egy átmeneti, csak az adott kapcsolatban élő adatbázist nyit; a duckdb.connect("northstar.duckdb") fájlba írja az adatot. Az explicit kapcsolat azért fontos, mert így nem a DuckDB rejtett alapértelmezett kapcsolatát használjuk, és mi irányítjuk, mely Python-objektumok melyik kapcsolatba regisztrálódnak.

CSV/Parquet scan. A read_parquet és read_csv_auto táblafüggvények a fájlt közvetlenül, pandas köztes DataFrame nélkül olvassák be a lekérdezésbe. A Parquet a fájlban tárolt sémát használja; CSV-nél a DuckDB mintavételezéssel becsli a típusokat. Az orders_raw.csv hiányzó distance_km értékei például NULL-ként jelennek meg, nem nullává alakított távolságként. A scan nem tölti be az egész fájlt a Python heapbe: a motor a saját pufferében dolgozik, és csak a kért oszlopokat, illetve a szükséges sorokat olvassa.

Relation. A con.sql("SELECT ...") hívás nem feltétlenül anyagiasítja azonnal az eredményt: egy DuckDBPyRelation objektumot ad vissza. Ez a reláció a lekérdezés terve, amely a relációs algebra kifejezéseként írja le, mit szeretnénk számolni. A tényleges végrehajtás akkor történik, amikor az eredményt lekérjük, például .fetchall(), .df(), .arrow() vagy .show() hívással. A Relation láncolható is: szűrés, vetítés és aggregálás építhető rá, hasonlóan ahhoz, ahogy a Polars lazy tervét fogjuk a következő alfejezetben használni.

Ha csak a séma érdekel, a DESCRIBE SELECT ... lekérdezés vagy a Relation oszlopneveinek vizsgálata nem anyagiasítja a teljes eredményt.

Szál- és memóriakeret. A DuckDB a folyamaton belül több szálat használhat egy lekérdezés párhuzamos végrehajtására. A SET threads TO 4 felső korlátot ad a szálhasználatra, a SET memory_limit='2GB' pedig a motor puffermemóriáját korlátozza. Ez a korlát nem foglalja magában a Python oldali DataFrame-eket: ha rel.df()-et hívunk, a pandas másolat további heapmemóriát fogyaszt. A fájlalapú .duckdb állomány egyszerre jellemzően egy író folyamatnak való; többfelhasználós írási tranzakciókra nem ez a megfelelő eszköz.

8.2.3 Teljes minimális munkapélda

Két apró kézi táblával ellenőrizzük a connect, a regisztráció, a Relation és a konverzió együttműködését. Futtatás a projektgyökérből: uv run python mini_duckdb.py (Linux/macOS terminál; PowerShellben ugyanez a uv parancs, csak az útvonalírás térhet).

import pandas as pd
import duckdb

mini_orders = pd.DataFrame({
    "order_id": ["A", "B"],
    "shipping_fee_minor": [300, 0],
})
mini_items = pd.DataFrame({
    "order_id": ["A", "A", "B"],
    "line_total_minor": [1000, 2000, 500],
})

con = duckdb.connect(":memory:")
con.register("mini_orders", mini_orders)
con.register("mini_items", mini_items)

rel = con.sql("""
SELECT
    (SELECT count(*) FROM mini_orders)          AS n_orders,
    (SELECT count(*) FROM mini_items)           AS n_items,
    (SELECT sum(shipping_fee_minor)
       FROM mini_orders)                        AS shipping_ok,
    (SELECT sum(o.shipping_fee_minor)
       FROM mini_items i
       JOIN mini_orders o USING (order_id))     AS shipping_dup
""")

print(type(rel).__name__)
df = rel.df()
print(df.to_string(index=False))
print("df.shape =", df.shape)

Az utolsó sort a szövegíró nem futtatta; a kézi számításból következik. A shipping_dup azért 600, mert a három tételsor közül az A rendelés kétszer szerepel, és a 300 centes szállítási díj kétszer adódik az összeghez.

8.2.4 Eredmény értelmezése

A kis példa a Join-kardinalitás és duplázódó célváltozó join-kardinalitási leckéjét ismétli: a belső join sorszáma a tételek számával egyezik, ezért a rendelésszintű oszlopok ismétlődnek. A mutatók definíciója skalár eredményt ad egy sorban:

A joined sorból számolt

M_3' = \sum_{k=1}^{3} s_{o(k)} = 300+300+0 = 600

nem ugyanaz a mutató. Az IDE, típusinformáció és hibakeresés alfejezetben tanult debuggeres shape-ellenőrzés itt is hasznos: a rel.df().shape legyen (1, 4), mert egy aggregált sor négy skalár oszlopot tartalmaz. Mielőtt nagy fájlra váltunk, ezt a kézi kontrollt érdemes megismételni.

A laborban a Northstar fájlokkal dolgozunk. A DuckDB-függőséget először a projektbe kell venni:

# Linux/macOS terminál, a lab projekt gyökérkönyvtárából
uv add duckdb
uv lock
uv sync --locked

Ezután a DuckDB közvetlen Parquet-scanjeit összehasonlítjuk egy SQLite kontrollal. A kód a szövegírás során nem futott; a nyomtatott pénzösszegek a helyi manifestből adódnak, ezért itt nem adunk előre konkrét számot.

import sqlite3
import time

import pandas as pd
import duckdb

t0 = time.perf_counter()

# DuckDB: közvetlen fájlból, explicit kapcsolattal.
con = duckdb.connect(":memory:")
con.execute("SET threads TO 4")
con.execute("SET memory_limit='2GB'")

duck_sql = """
SELECT
    (SELECT count(*) FROM read_parquet('data/orders.parquet'))  AS n_orders,
    (SELECT count(*) FROM read_parquet('data/order_items.parquet')) AS n_items,
    (SELECT sum(basket_minor) FROM read_parquet('data/orders.parquet')) AS basket_sum,
    (SELECT sum(shipping_fee_minor) FROM read_parquet('data/orders.parquet')) AS shipping_sum,
    (SELECT sum(i.line_total_minor)
       FROM read_parquet('data/order_items.parquet') i
       JOIN read_parquet('data/orders.parquet') o
         ON i.order_id = o.order_id) AS line_total_joined
"""
duck_rel = con.sql(duck_sql)
duck_result = duck_rel.df().iloc[0].to_dict()
t_duck = time.perf_counter() - t0

# SQLite kontroll: a CSV-ket pandas betölti, az stdlib sqlite3 végrehajtja.
sqlite_con = sqlite3.connect(":memory:")
orders = pd.read_csv("data/orders.csv", parse_dates=["order_ts"])
items = pd.read_csv("data/order_items.csv")
orders.to_sql("orders", sqlite_con, index=False)
items.to_sql("order_items", sqlite_con, index=False)

sqlite_sql = """
SELECT
    (SELECT count(*) FROM orders)                                     AS n_orders,
    (SELECT count(*) FROM order_items)                                AS n_items,
    (SELECT sum(basket_minor) FROM orders)                            AS basket_sum,
    (SELECT sum(shipping_fee_minor) FROM orders)                      AS shipping_sum,
    (SELECT sum(i.line_total_minor)
       FROM order_items i
       JOIN orders o ON i.order_id = o.order_id)                      AS line_total_joined
"""
sqlite_row = sqlite_con.execute(sqlite_sql).fetchone()
sqlite_result = dict(zip(
    ["n_orders", "n_items", "basket_sum", "shipping_sum", "line_total_joined"],
    sqlite_row))
t_total = time.perf_counter() - t0

print(duck_result)
print(sqlite_result)
print(f"duckdb_wall_s={t_duck:.3f}, total_wall_s={t_total:.3f}")

for key in sqlite_result:
    duck_val = duck_result[key]
    sqlite_val = sqlite_result[key]
    if sqlite_val is None:
        assert duck_val is None, key
    else:
        assert int(duck_val) == int(sqlite_val), key
print("kontroll: a DuckDB és SQLite mutatók egyeznek")

Ha még nincs Parquet-állomány, a CSV-től Parquet-ig: séma és partíció alapján készítsük el, vagy ideiglenesen cseréljük a read_parquet('data/orders.parquet') kifejezéseket read_csv_auto('data/orders.csv')-re. A n_orders=2400 és n_items=4796 a szerződésből várható; a basket_sum és a line_total_joined egyezése a tételszintű kosárösszeg invariant, egész minor unitban. A duckdb_wall_s egyszeri, a saját gépen mért tájékoztató idő; teljesítmény-összehasonlításhoz gép, adatméret, ismétlésszám és azonos input kellene.

8.2.5 Feladatok

con.register("orders", pd.read_csv("data/orders.csv"))
con.register("order_items", pd.read_csv("data/order_items.csv"))
bad = con.sql("""
SELECT sum(o.shipping_fee_minor) AS shipping
FROM order_items i JOIN orders o ON i.order_id = o.order_id
""").df()

A kapott shipping nagyobb, mint a SELECT sum(shipping_fee_minor) FROM orders eredménye. A Join-kardinalitás és duplázódó célváltozó join-kardinalitási szabályai alapján magyarázd meg, miért, és adj egy javított lekérdezést.

8.2.6 Megoldások és indoklás

SELECT sum(shipping_fee_minor) AS shipping FROM orders

Vagy előaggregálással egy rendelésenként egy reprezentáns sort kell venni, például (order_id, shipping_fee_minor) egyedi párokból.

SELECT
    count(*) AS n_orders,
    (SELECT count(*)
       FROM order_items i
       JOIN orders o ON i.order_id = o.order_id
      WHERE o.priority_flag = 1) AS n_items,
    sum(shipping_fee_minor) AS shipping_sum
FROM orders
WHERE priority_flag = 1

Az invariáns teljesül, mert a belső select a priority rendelésekhez tartozó tételeket számolja, a külső WHERE pedig a rendelésszintű shipping_fee_minor összeget szűkíti ugyanarra a halmazra. A szűrt shipping_sum így nem tartalmazza a priority_flag = 0 rendelések díját.

8.2.7 Gyakori tévedés és ellenpélda

A beágyazott analitikus adatbázis nem minden többfelhasználós tranzakciós szerver helyettesítője. Ha két párhuzamos folyamat ugyanarra a .duckdb fájlra ír, az egyik zárolási hibát vagy sérült állapotot kaphat; egy szerveres OLTP-rendszer, például egy PostgreSQL, a tranzakciókat, a felhasználókezelést és az egyidejű írásokat más mechanizmussal kezeli. A DuckDB-t ezért kutatói folyamaton belüli, fájlok feletti analitikus lekérdezésre használjuk, nem alkalmazásszervernek. A Join-kardinalitás és duplázódó célváltozó alfejezetből ismert, hogy a join előtti kulcsegyediség és a join utáni sorváltozás ellenőrzése akkor is kötelező, ha a motor gyors. Önellenőrzés: mielőtt új adatforrásra váltanál, futtasd a mini példát, és győződj meg róla, hogy a shipping_dup pontosan kétszerese a shipping_ok-nak.

Rendelésszintű aggregációnál a rendeléshez tartozó shipping_fee_minor összege megsokszorozódhat a tételsorokkal való összekapcsolás után. A tételsorokat előbb order_id szerint aggregáljuk, és csak azután kapcsoljuk az egyedi rendelésekhez.

8.2.8 Rövid összefoglaló

A DuckDB beágyazott analitikus SQL-motort ad: explicit duckdb.connect kapcsolattal, szerver nélkül olvasunk CSV-t és Parquet-et. A con.sql Relation objektumot ad, amely a lekérdezési terv; a tényleges anyagiasítást a .df(), .fetchall() vagy .arrow() hívás váltja ki. A fájlscan elkerüli a pandas köztes DataFrame-et, de a .df()-felé konvertálás a Python heapben további memóriát fogyaszt. A Northstar mutatók DuckDB-vel és SQLite-tal azonos SQL-definícióval számolhatók; a kontroll a join-kardinalitás miatt a rendelésszintű oszlopokra külön figyelmet igényel. A következő alkalmazás: a Polars lazy tervének és a DuckDB Relation viselkedésének összehasonlítása.

8.2.1 Cél és konkrét helyzet