8.3 Polars lazy terv és streaming korlát

Ez az alfejezet bővítő egység: a tanulási főút pandas + SQL marad, de a Polars lazy terve fontos ellenpont, mert a tervépítést és a végrehajtást különválasztja.

8.3.1 Cél és konkrét helyzet

Egy data scientist a Northstar laborban ügyfélenkénti, a 2025. május 1. előtti rendelésekből számolt tétel-összértéket akar előállítani. Bemenet a lab/data/orders.csv (2400 rendelési sor) és a lab/data/order_items.csv (4796 tételsor). A megfigyelési egység itt a rendelési tétel; az aggregáció egysége a customer_id. A kimenet fájlba írt Parquet-tábla, oszlopai: customer_id, item_total_minor, item_rows, order_count. Az elvárt ellenőrzés: a tételösszegek összege megegyezik a cutoff előtti rendelések basket_minor értékeinek összegével, és a kimenet sorainak száma legfeljebb 300, mert ennyi ügyfél van a customers.csv-ben.

Mielőtt kódot írsz, adj durva becslést: a rendelések 2025 első 180 napjára esnek, a cutoffig 120 nap telik el, ezért egyenletes eloszlást feltételezve körülbelül 2400 * 120/180 = 1600 rendelés lehet a cutoff előtt; a pontos számot a futtatás adja. Az ellenőrző invariant: sum(item_total_minor) == sum(basket_minor) a leszűrt rendelésekre.

8.3.2 Lusta terv, collect és sink

A Polars kifejezéseiben a pl.scan_csv(...) vagy pl.scan_parquet(...) nem olvas be adatot. Egy pl.LazyFrame objektumot ad, amely leírja a forrást és az elvégzendő lépéseket. A select, filter, group_by és join hívások csak új csomópontokat fűznek ehhez a tervhez; a tényleges beolvasást és számítást a collect() hívás váltja ki. A collect() eredménye egy hagyományos pl.DataFrame a Python heapben.

A scan_parquet ugyanezt a felületet adja Parquet-forrásra. A read_parquet ezzel szemben azonnal DataFrame-et ad, tehát eager módon betölti a fájlt. A sink_parquet a lazy tervet közvetlenül Parquet-fájlba anyagiasítja, anélkül hogy a teljes eredményt DataFrame-ként a heapben tartaná; ez akkor hasznos, ha a kimenet következő lépésben egy másik eszköz bemenete. Az explain() nem futtat semmit, csak a terv szöveges formáját mutatja.

A Polars lazy motorja a terv optimalizálásakor predikátum- és vetületletolást végez. A filter(pl.col("order_ts") < cutoff) a forrás közelébe, a select a lehető legkorábbra kerülhet, így a feldolgozó kevesebb sort és oszlopot mozgat. A streaming motor ezen felül akkor is haladhat batch-ekben, ha a teljes bemenet nem fér a memóriába, de a csoportosítás állapota továbbra is adatfüggő; ezt a gyakori tévedés részben fejtjük ki.

8.3.3 Szűrés korábbra hozása: algebrai azonosság

Jelölje $O$ a rendelések tábláját, $I$ a tételek tábláját, $\bowtie$ az order_id-re vett inner joint, $\sigma_p$ pedig a $p$ predikátummal történő sorszűrést. Ha $p$ csak az $O$ oszlopait használja, akkor

$\sigma_{o.order\_ts < c}(O \bowtie I) \equiv \sigma_{o.order\_ts < c}(O) \bowtie I,$

ahol $c$ a cutoff időbélyeg, $o.order\_ts$ a rendelés időpontja. A bal oldalon előbb a join, utána a szűrés; a jobb oldalon a szűrés a join elé kerül. A két oldal sorai azonosak, de a közbülső méret különbözik: a bal oldal közbülső táblája minden rendelés összes tételsorát tartalmazza, a jobb oldalé csak a cutoff előtti rendelések tételeit.

Kis kézi példa: legyen $O$ három rendelés:

A: 2025-01-01, basket_minor=100
B: 2025-05-02, basket_minor=200
C: 2025-04-30, basket_minor=300

$I$ tételei: A-hoz 40 és 60, B-hez 200, C-hez 100 és 200. A join → filter út előbb 5 tételsort állít elő, majd a szűrés 4 sort hagy meg. A filter → join út már a szűrt 2 rendeléssel lép a joinba, és közvetlenül 4 tételsort ad. A tételösszeg mindkét esetben 400, azaz a cutoff előtti kosárösszegek összege 100 + 300. A Polars explain() kimenete várhatóan a FILTER lépést a JOIN elé helyezi, még akkor is, ha a kódban a join előtt csak a base filterét írtuk.

8.3.4 Teljes minimális példa a Northstar laborral

A labor CSV-t használ, ezért a kód scan_csv-t hív; a scan_parquet azonos felületet ad, ha a forrás Parquet. Telepítsd a bővítő csomagokat a labor környezetébe, majd futtasd a scriptet:

# munkakönyvtár: lab
uv add polars pyarrow
uv run python s031_example.py

A következő kód szemléltető: a kézirat összeállítása során nem futtatták, ezért a kimeneti értékek elméletileg vártak, nem mért futási naplók. A pontos számokat a helyi futtatás és a lab/manifest.json ellenőrzi.

# lab/s031_example.py
from pathlib import Path
import polars as pl

DATA = Path("data")
OUT = Path("artifacts")
OUT.mkdir(exist_ok=True)

CUTOFF = (
    pl.lit("2025-05-01T00:00:00+00:00")
    .str.to_datetime("%Y-%m-%dT%H:%M:%S%:z")
)


def build_customer_summary(
    orders: pl.LazyFrame, items: pl.LazyFrame
) -> pl.LazyFrame:
    base = (
        orders.filter(pl.col("order_ts") < CUTOFF)
        .select(["order_id", "customer_id", "basket_minor"])
    )
    return (
        base.join(items, on="order_id", how="inner")
        .group_by("customer_id")
        .agg(
            pl.col("line_total_minor").sum().alias("item_total_minor"),
            pl.len().alias("item_rows"),
            pl.col("order_id").n_unique().alias("order_count"),
        )
    )


def main() -> None:
    orders = pl.scan_csv(DATA / "orders.csv", try_parse_dates=True)
    items = pl.scan_csv(DATA / "order_items.csv")

    lazy_summary = build_customer_summary(orders, items)
    print(lazy_summary.explain())

    result = lazy_summary.collect()
    print(result.shape)

    # eager kontroll: ugyanaz a művelet azonnali DataFrame-ekkel
    orders_e = pl.read_csv(DATA / "orders.csv", try_parse_dates=True)
    items_e = pl.read_csv(DATA / "order_items.csv")
    result_e = build_customer_summary(orders_e.lazy(), items_e.lazy()).collect()
    assert result_e.equals(result), "eager és lazy eredmény nem egyezik"

    order_total = (
        orders.filter(pl.col("order_ts") < CUTOFF)
        .select(pl.col("basket_minor").sum())
        .collect()
        .item()
    )
    item_total = result.select(pl.col("item_total_minor").sum()).item()
    assert order_total == item_total, (order_total, item_total)
    print(f"invariant ok: {item_total}")

    lazy_summary.sink_parquet(OUT / "s031_customer_summary.parquet")
    reloaded = pl.read_parquet(OUT / "s031_customer_summary.parquet")
    assert reloaded.equals(result)

    streamed = lazy_summary.collect(streaming=True)
    assert streamed.equals(result)


if __name__ == "__main__":
    main()

A %:z formátum a +00:00 időeltolódást értelmezi a cutoff kifejezésben. A try_parse_dates=True hatására az order_ts Datetime típusú lesz; ha a telepített Polars szerint nem ismeri fel a mezőt, a hibaüzenet ezt jelzi, és a datetimeszerződést külön ellenőrizni kell. A result.shape várhatóan (M, 4), ahol M <= 300; a pontos M a futtatásból derül ki.

A kód a Polars 1.x API-ját feltételezi: group_by, sink_parquet, collect(streaming=True). A telepített verziót a uv.lock rögzíti; ha más verzió települ, a Polars lazy útmutató az ellenőrzési kiindulópont.

8.3.5 Eredmény értelmezése és ellenőrzése

A lazy explain nem bizonyíték arra, hogy a szűrés korábbra került; a tényleges tervet a Polars saját kimenete mutatja. A numerikus kontrollt a fenti invariant adja: a szűrt rendelések basket_minor összege megegyezik a tételösszegek összegével. Ez egy one-to-many join után sem duplázódik, mert az összegzett oszlop a tételtáblából származik, a basket_minor pedig csak a kontrollhoz, a rendelés egy-egy sorából kerül összegzésre.

A reloaded.equals(result) a Parquet-visszaolvasás sémáját és értékeit ellenőrzi. Az eager és lazy eredmény egyezése azt mutatja, hogy a lazy terv nem változtatta meg a szemantikát, csak a végrehajtás sorrendjét. A memóriaigény itt nem mért: a pandasos és eager DataFrame-ek a teljes CSV-t betöltik, míg a lazy scan elvileg folyamatosan olvashat, de ennek igazolásához valós tracemalloc vagy peak-RSS mérés, rögzített gép, adatméret és ismétlésszám szükséges; ezt a feladatban kérjük számon.

8.3.6 Feladatok

Értelmezés. Magyarázd el, mi a különbség a pl.scan_parquet("orders.parquet") és a pl.read_parquet("orders.parquet") hívás között. Melyik hívás vált ki tényleges fájl-olvasást? Hogyan ellenőriznéd a típusukból, hogy a terv még nem futott?
Implementálás. Készíts lazy tervet, amely a 2025. április 1. előtti rendelésekre régiónként kiszámítja az egyedi rendelések számát és a tétel-összértéket. Bemenet: lab/data/orders.csv, lab/data/order_items.csv, lab/data/customers.csv. A kimenetet írd a lab/artifacts/s031_region.parquet fájlba, és ellenőrizd, hogy a visszaolvasott táblának legfeljebb 3 sora van, a régiók halmaza pedig {"north", "central", "south"}.

Hibadiagnózis. Adott a következő tervrészlet:

wrong = (
    orders.filter(pl.col("order_ts") < CUTOFF)
    .join(items, on="order_id", how="inner")
    .group_by("order_id")
    .agg(
        pl.col("line_total_minor").sum().alias("item_total_minor"),
        pl.col("shipping_fee_minor").sum().alias("shipping_fee_minor"),
    )
)

Miért hibás az így számolt shipping_fee_minor? Javítsd a tervet úgy, hogy a szállítási díj rendelésenként pontosan egyszer szerepeljen, és a line_total_minor összege továbbra is tételszintről származzon.

Megváltozott feltétel. Tegyük fel, hogy a customers.csv 10 millió ügyfelet tartalmaz, és ugyanezt a group_by("customer_id") aggregációt futtatod lazy módon. Miért nem garantálja a lazy API önmagában az állandó memóriaigényt? Nevezz meg legalább egy műveletet vagy adatfüggő állapotot, amely a memóriát növelheti, és írd le, milyen méréssel döntenéd el, hogy a sink_parquet vagy a collect a megfelelő kimeneti forma.

8.3.7 Megoldások és indoklás

A scan_parquet pl.LazyFrame-et ad, amely a fájlt csak tervként hivatkozza; a read_parquet pl.DataFrame-et ad, tehát azonnal beolvassa a fájlt. A tényleges I/O-t a lazy esetben a collect() váltja ki. A típusellenőrzés: isinstance(obj, pl.LazyFrame) igaz a scan, hamis a read esetén; a collect() után a típus pl.DataFrame. Az explain() nem olvas fájlt, így önmagában nem bizonyíték a futtatásra, de megmutatja a tervezett lépéseket.

Egy lehetséges megoldás:

cutoff = (
    pl.lit("2025-04-01T00:00:00+00:00")
    .str.to_datetime("%Y-%m-%dT%H:%M:%S%:z")
)

orders = pl.scan_csv("data/orders.csv", try_parse_dates=True)
items = pl.scan_csv("data/order_items.csv")
customers = pl.scan_csv("data/customers.csv")

pre = orders.filter(pl.col("order_ts") < cutoff)

with_region = (
    pre.join(
        customers.select(["customer_id", "region"]),
        on="customer_id",
        how="inner",
    )
    .select(["order_id", "region"])
)

summary = (
    with_region.join(items, on="order_id", how="inner")
    .group_by("region")
    .agg(
        pl.col("order_id").n_unique().alias("order_count"),
        pl.col("line_total_minor").sum().alias("item_total_minor"),
    )
    .sort("region")
)

summary.sink_parquet("artifacts/s031_region.parquet")

reloaded = pl.read_parquet("artifacts/s031_region.parquet")
assert set(reloaded["region"].to_list()) == {"north", "central", "south"}
assert len(reloaded) <= 3

# invariant kontroll
order_count_expected = (
    pre.select(pl.col("order_id").n_unique()).collect().item()
)
assert reloaded["order_count"].sum() == order_count_expected

item_total_expected = (
    pre.join(items, on="order_id", how="inner")
    .select(pl.col("line_total_minor").sum())
    .collect()
    .item()
)
assert reloaded["item_total_minor"].sum() == item_total_expected

Az invariáns: a regionális rendelésszámok összege a cutoff előtti egyedi rendelések száma, a tételösszegek összege pedig a cutoff előtti tételek line_total_minor összege.

A hiba oka: a shipping_fee_minor rendelésenként egyetlen érték, de a join után minden tételsorban ismétlődik, ezért a pl.col("shipping_fee_minor").sum() a rendelés tételszámával megszorozza az értéket. Ez ugyanaz a one-to-many csapda, amelyet a szállítási díj kapcsán a 4.3 modulban már elkerültünk. A javítás: a tételösszeget külön csoportosítással állítsd elő, majd joinold a rendelésszintű oszlopokhoz:
```
fixed = (
    orders.filter(pl.col("order_ts") < CUTOFF)
    .join(items, on="order_id", how="inner")
    .group_by("order_id")
    .agg(pl.col("line_total_minor").sum().alias("item_total_minor"))
    .join(
        orders.filter(pl.col("order_ts") < CUTOFF)
        .select(["order_id", "shipping_fee_minor"]),
        on="order_id",
        how="inner",
    )
)
```
Itt a shipping_fee_minor már a rendelés egy-egy sorából kerül a joinba, így minden order_id-hoz pontosan egyszer jelenik meg.
A group_by("customer_id") aggregáció a csoportok kulcsait és a részösszegeket hash-táblában tartja. Ha a kulcsok száma 10 millió, az állapot mérete a csoportok számával arányos, ezért a csúcsigény meghaladhatja a rendelkezésre álló RAM-ot; a lazy API ettől nem véd. A sink_parquet a kimenetet közvetlenül fájlba írhatja, így a Python heapben nem kell a teljes eredmény-DataFrame-et tartani, de a csoportállapot akkor is adatfüggő marad. Megfelelő döntéshez mérd a peak RSS-t vagy tracemalloc-ot, rögzítsd a gép, az input méretét és az ismétlésszámot, és hasonlítsd össze a collect és sink_parquet futását ugyanazon a terhelésen.

8.3.8 Gyakori tévedés és önellenőrzés

A leggyakoribb tévedés: „ha a kód lazy, akkor automatikusan streaming, tehát alacsony a memóriaigény.” Ez hamis. A lazy API csak a tervépítést választja el a végrehajtástól; a memóriát a futtatandó művelet és az adatfüggő állapot határozza meg. A group_by sok csoportkulccsal, a nem streaming módon futó rendezés vagy egy nagy join rács akkor is növelheti a csúcsigényt, ha a terv explain() kimenete szép. Ellenpélda: 10 millió egyedi customer_id esetén a group_by(...).agg(pl.len()) csoportállapota önmagában nagy lehet, miközben a terv ugyanúgy néz ki, mint a 300 ügyfelet tartalmazó labor esetében.

Ez rokon a 7.3 CTE-s ablakfüggvényes példájával: a PARTITION BY customer_id és a ROW_NUMBER is fenntart rendezett állapotot, a tervsorrend nem törli az adatfüggő költséget. A 4.3 modulhatár-elve itt is érvényes: a build_customer_summary függvényben rögzített definíciót a notebook, a script és a CLI ugyanúgy hívhatja; a lazy terv nem teszi feleslegessé a tiszta függvényhatárt.

Önellenőrzés: (1) meg tudom mondani, melyik hívás ad LazyFrame-et és melyik DataFrame-et? (2) be tudom bizonyítani a mini példán, hogy a filter a join elé hozható? (3) meg tudom nevezni, miért nem duplázódik a basket_minor a példában? (4) meg tudom nevezni egy olyan műveletet, amelynél a lazy forma sem garantál állandó memóriát?

A lazy végrehajtás nem jelent állandó vagy garantáltan alacsony memóriaigényt. A nagy kardinalitású group_by, a rendezés és egyes joinok jelentős köztes állapotot igényelhetnek; a streaming-végrehajtás támogatása a konkrét művelettől és könyvtárverziótól is függ.

8.3.9 Rövid összefoglaló

A scan_parquet és scan_csv LazyFrame-et ad: tervet, nem adatot. A select, filter és group_by csak tervcsomópontokat fűz; a collect anyagiasít DataFrame-et, a sink_parquet közvetlenül fájlba ír. A predikátum-letolás miatt a cutoff-szűrés a join elé kerülhet, ami algebrailag azonos eredményt ad kevesebb közbülső sorral. A streaming motor csökkentheti a heap-csúcsot, de a csoportállapot a kulcsok számával arányos; a lazy API nem garancia korlátlan memóriára. A következő lépés az II. ellenőrző pont: adatból ellenőrzött feature-tábla ellenőrző pont: a pandas, DuckDB és Polars kimenetek azonos mutatódefinícióval történő együttes ellenőrzése.