34.1 Nagyobb mint RAM: előbb adatfolyam

A Northstar rendelési tábla a kezdőlaborban mindössze 2400 sor; a következő példában azt vizsgáljuk, mi történik, ha ugyanezt a sémát egy nagyítási paraméterrel több tízmillió sorra növeljük, és a gépünk RAM-ja nem nő együtt az adattal. A cél nem az, hogy azonnal klasztert vagy felhős erőforrást válasszunk, hanem hogy előbb megmérjük: mennyi adatot kell ténylegesen beolvasni, és milyen technikákkal csökkenthetjük a memória csúcsát egyetlen gépen.

34.1.1 Cél és konkrét helyzet

A feladatunk egy egyszerű üzleti kérdés: a Northstar rendelésekben mekkora az átlagos distance_km a priority_flag szerinti két csoportban? A bemenet a már megtisztított orders.csv, amelynek oszlopai között szerepel a distance_km (float, kilométer), a priority_flag (0/1 egész) és számos további, a kérdéshez nem szükséges mező (például actual_delivery_at, basket_minor, delivery_days). A megfigyelési egység egy rendelés. Az elvárt kimenet: soronként a priority_flag, a csoport elemszáma és az átlagos távolság; ellenőrzésként ugyanannak a két csoportnak az elemszáma összegezve a bemeneti sorok számát adja, és a csoportátlagok a teljes átlagot súlyozottan reprodukálják.

Előzetes becslés: ha a táblát scale_factor = 10 000-rel nagyítjuk, akkor $2400 \times 10\,000 = 24\,000\,000$ sorunk lesz. A teljes táblában 15 oszlop van; ha átlagosan oszloponként 8 bájtot számolunk (a dátum-szövegek ennél nagyobbak, a 0/1 flag kisebb), akkor a nyers reprezentáció nagyságrendje $24\,\text{M} \times 15 \times 8 \approx 2.9\,\text{GB}$. A tényleges memóriaigény erősen függ a tárolási formátumtól, a stringoszlopok kódolásától és a végrehajtó motortól; a 4–6 GB legfeljebb szemléltető becslés, nem általános RAM-követelmény. A kérdéshez viszont csak két oszlop kell: $24\,\text{M} \times 2 \times 8 = 384\,\text{MB}$, ami egy átlagos fejlesztői laptopon is kényelmesen elfér.

34.1.2 Fogalmi magyarázat: hat eszköz a beolvasás csökkentésére

Projection (oszlopkiválasztás). A vetítés azt jelenti, hogy a lekérdezésben vagy a beolvasó hívásban csak azokat az oszlopokat kérjük el, amelyekre a számításnak ténylegesen szüksége van. Az oszlopos (columnar) fájlformátumok, mint a Parquet, ezt hatékonyan támogatják: a motor egyszerűen nem olvassa be a többi oszlop adatblokkjait. Ha a distance_km és a priority_flag oszlopokat kérjük, a basket_minor, a dátumok és a többi mező bájtjai a lemezen maradnak. A memóriaigény így $M_{\text{read}} = n \sum_{j \in S} b_j$ bájt, ahol $n$ a sorok száma, $S$ a kiválasztott oszlophalmaz, $b_j$ pedig az egy sorra jutó bájtmennyiség az adott oszlopban.

Partition pruning (partícióvágás). Ha a fájl nem egyetlen nagy objektum, hanem könyvtárstruktúrába rendezett partíciókra van osztva — például year=2025/month=01/ —, akkor a lekérdezésben szereplő szűrési feltétel alapján a motor a partíciók egy részét teljesen kihagyja. Ha az adat $P$ egyenlő partícióra oszlik, és a feltétel $k$ partíciót érint, akkor a beolvasott adatmennyiség $\frac{k}{P}$-szeresére csökken. A vágás akkor működik, ha a partíciós oszlop értéke magából a könyvtárnévből kiolvasható, és a motor nem kényszerül minden fájl megnyitására.

Chunk (darab). A chunk-olvasás során a fájlt nem egyszerre, hanem rögzített méretű sorcsoportokban vagy row groupokban töltjük be. A Parquet eleve row groupokra tagolódik; a DuckDB, a Polars és a Pandas is képes batch-enként olvasni. Ha egyszerre $B$ soros darabot tartunk a memóriában, és az aggregációs állapot korlátos, akkor a memóriaigény $O(B)$ lehet; rendezésnél, nagy kardinalitású csoportosításnál vagy joinnál ez nem garantált.

Streaming. A streaming feldolgozás azt jelenti, hogy a számítás darabonként történik, és a közbenső állapot nem tartalmazza a teljes bemenetet. Egy csoportosított átlaghoz elég darabonként összegezni a distance_km értékeket és a sorok számát; a végeredményhez nincs szükség arra, hogy mind a 24 millió sor egyszerre a memóriában legyen. A két előre ismert csoportot használó streaming átlag számításának aggregációs állapota a sorok számában $O(1)$ (a beolvasási batch ezen felül memóriát igényel), míg a beolvasás utáni in-memory aggregáció $O(n)$.

Spill. Ha egy művelet — például rendezés vagy join — közbenső állapota meghaladja a beállított memóriakorlátot, a motor a felesleget lemezre írja. Ez a spill: a lekérdezés nem omlik össze, de lassabb lesz, mert lemez I/O keletkezik. Egy külső rendezésnél a lemezre írt futamok száma közelítőleg $\lceil M_{\text{work}} / M_{\text{limit}} \rceil$, ahol $M_{\text{work}}$ a rendezendő adat memóriaigénye, $M_{\text{limit}}$ pedig a rendelkezésre álló munkamemória. A spill tehát biztonsági szelep, nem gyorsítás.

Small files. Ha a partíciók túl aprók, minden fájl megnyitása, metaadat-olvasása és lezárása fix költséget jelent. Sok ezer kis fájl esetén ez a fix költség nagyobb lehet, mint maga az adatolvasás. Ökölszabály: egy partíciófájl ne legyen néhány tíz MB-nál kisebb, és a row group mérete legalább százezres nagyságrendű sor legyen. A partícióvágás tehát nem jelenti azt, hogy tetszőlegesen finomra kell darabolni az adatot.

A teljes beolvasás és a vetített streaming közötti különbség algoritmikus értelemben is megfogható: mindkét út időkomplexitása $O(n)$, de a konstans szorzó és a memóriaigény nagyon eltérő. A vetítés a beolvasott bájtok számát csökkenti, a streaming pedig a memóriában tartott állapotot.

34.1.3 Teljes minimális munkapélda

A példát a Northstar lab/orders.csv fájljából kiindulva építjük fel. A nagyításhoz DuckDB-t használunk, mert a Parquet-írás során explicit row group méretet adhatunk, és a lekérdezésekben a projection és a pruning automatikusan érvényesül. A kód a lab/ könyvtárból indul, a generált nagyított fájl a data/ alkönyvtárba kerül.

# data/generate_large.py
# A nagyított tanulótábla előállítása; a scale_factor a nagyítási paraméter.
import duckdb

con = duckdb.connect("data/northstar_large.duckdb")
scale_factor = 100  # 2400 * 100 = 240 000 sor; növelhető, pl. 10000

con.execute("""
    COPY (
        SELECT
            o.order_id || '-' || r.i AS order_id,
            o.customer_id,
            o.order_ts,
            o.promised_days,
            o.quantity_sum,
            o.item_rows,
            o.basket_minor,
            o.shipping_fee_minor,
            o.currency,
            o.distance_km,
            o.priority_flag,
            o.delivery_days,
            o.actual_delivery_at,
            o.label_available_at,
            o.target_late
        FROM read_csv('lab/orders.csv', header = true) AS o
        CROSS JOIN (SELECT i FROM range(0, ?) AS r(i)) AS r
    ) TO 'data/orders_large.parquet'
      (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 100000)
""", [scale_factor])

print(con.execute("""
    SELECT count(*) AS n, count(DISTINCT order_id) AS n_unique
    FROM read_parquet('data/orders_large.parquet')
""").fetchall())

A generálás után két utat hasonlítunk össze. Az első a vetített aggregáció: a lekérdezésben csak a két szükséges oszlop szerepel, így a Parquet-olvasó a többi oszlopot nem is érinti.

# data/compare_reads.py
import duckdb
import time

con = duckdb.connect("data/northstar_large.duckdb")

# 1. út: vetített aggregáció
t0 = time.perf_counter()
result_projected = con.execute("""
    SELECT priority_flag,
           count(*)             AS n,
           avg(distance_km)     AS avg_km
    FROM read_parquet('data/orders_large.parquet')
    GROUP BY priority_flag
    ORDER BY priority_flag
""").fetchall()
t1 = time.perf_counter()

# 2. út: teljes beolvasás, majd aggregáció a memóriában
con.execute("DROP TABLE IF EXISTS full_orders")
t2 = time.perf_counter()
con.execute("""
    CREATE TEMP TABLE full_orders AS
    SELECT * FROM read_parquet('data/orders_large.parquet')
""")
result_full = con.execute("""
    SELECT priority_flag,
           count(*)             AS n,
           avg(distance_km)     AS avg_km
    FROM full_orders
    GROUP BY priority_flag
    ORDER BY priority_flag
""").fetchall()
t3 = time.perf_counter()

print("Vetített eredmény: ", result_projected)
print("Teljes beolvasás:  ", result_full)
print(f"Vetített aggregáció idő: {t1 - t0:.3f} s")
print(f"Teljes beolvasás + aggregáció idő: {t3 - t2:.3f} s")

A fenti kódot a kapcsolódó példaprojektben még nem ellenőriztük, ezért a konkrét időértékek szemléltető jellegűek. Az elméletileg várt viselkedés: a vetített változat kevesebb bájtot olvas a lemezről, és a memória-csúcs is alacsonyabb; a teljes beolvasásnál a full_orders ideiglenes tábla a 15 oszlop minden értékét a RAM-ba teszi. Kis scale_factor mellett a különbség alig látszik, mert az I/O-időt a parancsértelmezés fix költsége dominálja; a különbség a nagyított adaton válik mérhetővé.

Ugyanez a feladat Polars streaminggel is kifejezhető. Itt a scan_parquet lusta (lazy) tervet ad, a select végzi a vetítést, a collect(engine="streaming") pedig streaming aggregációt kér:

import polars as pl

lf = (
    pl.scan_parquet("data/orders_large.parquet")
    .select(["priority_flag", "distance_km"])
    .group_by("priority_flag")
    .agg([pl.count(), pl.col("distance_km").mean()])
)
result = lf.collect(engine="streaming")
print(result)

34.1.4 Eredmény értelmezése és kézi kontroll

A két út eredményének numerikusan egyeznie kell. A priority_flag két értéke (0 és 1) szerinti csoportokra a count(*) összege a bemeneti sorok száma: $n_0 + n_1 = 2400 \times \text{scale\_factor}$. A csoportátlagokból a teljes átlag súlyozottan visszaszámolható:

$$\bar{x} = \frac{n_0 \bar{x}_0 + n_1 \bar{x}_1}{n_0 + n_1}.$$

Ha a priority_flag eloszlása a nagyítás során nem változik — a cross join minden eredeti sort ugyanannyiszor másol —, akkor a csoportarányok és az átlagok is megőrződnek. Ez a tulajdonság a nagyítási eljárás invariantja: a 2400 soros alap és a nagyított tábla csoportátlagai megegyeznek. A gyakorlatban a lebegőpontos összegzés sorrendje miatt az utolsó biten eltérés lehet; a kontrollhoz $10^{-9}$ relatív tolerancia elegendő.

A vetített és a teljes beolvasás közötti idő- és memóriakülönbséget a scale_factor növelésével lehet mérhetővé tenni. A méréshez ugyanazon a gépen, azonos seed és fájl mellett, több ismétléssel kell dolgozni; a perf_counter egyszeri lefutása csak nagyságrendi tájékozódásra való. Ha a folyamat RAM-csúcsát pontosan akarjuk ismerni, a resource modul getrusage értéke vagy a DuckDB memória-statisztikája használható, nem a Parquet fájl mérete.

34.1.5 Feladatok

1. Értelmezés. Egy 200 millió soros táblában 12 oszlop van: 8 oszlop float64, 2 oszlop int32, 2 oszlop átlagosan 20 karakteres szöveg. Becsüld meg a nyers reprezentáció méretét bájtban (a szövegnél 20 bájt/sor értékkel számolj), majd a DataFrame overheaddel együtt, ha az overhead szorzó 1.8×. Hány GB RAM kell a teljes beolvasáshoz?

2. Implementálás. Írj DuckDB lekérdezést, amely a data/orders_large.parquet fájlból csak a distance_km és a priority_flag oszlopokat olvassa be, és csoportonként visszaadja az átlagot, a minimumot és a maximumot. Ellenőrizd, hogy a min(distance_km) <= avg(distance_km) <= max(distance_km) mindkét csoportban teljesül.

3. Hibadiagnózis. Egy kolléga a következő kódot írja a nagyított táblára:

   df = pd.read_parquet('data/orders_large.parquet')
   result = df[df['region'] == 'north'].distance_km.mean()

   Nevezz meg legalább három memóriaproblémát ebben a megoldásban, és javasolj javítást.

4. Megváltozott feltétel. Tegyük fel, hogy a nagyított táblát hónap szerint particionáljuk: year=2025/month=01/, year=2025/month=02/, és így tovább, 12 egyenlő partícióra. A lekérdezés csak a 2025. januári rendeléseket érinti. Mennyivel csökken a beolvasott adatmennyiség, ha a partíciós oszlopra szűrünk? Miért lehet mégis rossz, ha a partíciókat naponként hozzuk létre?

5. Továbbvitel. Egy rendezési feladat 5 GB munkamemória-igényű adatot dolgoz fel, de a motornak csak 1 GB spill-korlátot adtunk. Hány lemezfutamra számíthatsz közelítőleg? Hogyan változik a válasz, ha a korlátot 2 GB-ra emeljük?

34.1.6 Megoldások és indoklás

1. megoldás. A nyers reprezentáció soronként:

$$b = 8 \times 8 + 2 \times 4 + 2 \times 20 = 64 + 8 + 40 = 112 \text{ bájt/sor}.$$

A teljes nyers méret:

$$M_{\text{raw}} = 200\,000\,000 \times 112 = 22.4 \times 10^9 \text{ bájt} = 22.4 \text{ GB}.$$

A DataFrame overheaddel:

$$M_{\text{mem}} = 1.8 \times 22.4 = 40.3 \text{ GB}.$$

Tehát legalább 32–40 GB RAM szükséges a teljes beolvasáshoz; a tömörített Parquet fájl ennél jóval kisebb lehet, de ez nem csökkenti a memóriaigényt. Ha csak például 3 oszlopra van szükség, a vetített nyers méret $200\,\text{M} \times 24 = 4.8$ GB, ami 8.6 GB RAM-mal már kezelhető.

2. megoldás. A lekérdezés:

SELECT priority_flag,
       avg(distance_km) AS avg_km,
       min(distance_km) AS min_km,
       max(distance_km) AS max_km
FROM read_parquet('data/orders_large.parquet')
GROUP BY priority_flag
ORDER BY priority_flag;

Az ellenőrzés: minden visszaadott sorra igaz, hogy min_km <= avg_km <= max_km. A vetítés itt automatikus: a DuckDB a Parquet-ból csak a két hivatkozott oszlopot olvassa, mert a lekérdezésben más oszlop nem szerepel. A GROUP BY streaming aggregációként futhat, így a 24 millió soros bemenet sem kerül egyszerre a memóriába.

3. megoldás. A hibák: (a) a pd.read_parquet columns paraméter nélkül mind a 15 oszlopot beolvassa, pedig csak kettőre van szükség; (b) a region szűrés a betöltés után történik, tehát a motor nem tud partícióvágást végezni, és a szükségtelen sorokat is beolvassa; (c) a pandas a teljes DataFrame-t a RAM-ba teszi, így a memóriaigény a legrosszabb eset. Javítás:

df = pd.read_parquet(
    'data/orders_large.parquet',
    columns=['region', 'distance_km'],
    filters=[('region', '==', 'north')],
)
result = df['distance_km'].mean()

A filters paraméter a PyArrow motorral partícióvágást is végezhet, ha a partíciós oszlop a feltétel szerint van kialakítva; a columns paraméter a vetítést adja meg. Ha a fájl nem particionált, a szűrés legalább a betöltött oszlopokra korlátozódik, és a memória már csak két oszlopra nő.

4. megoldás. Ha a 12 hónap partíciói egyenletesek, akkor a januári szűrés $\frac{1}{12}$-ére csökkenti a beolvasott adatmennyiséget. A naponkénti partíciók esetén 365 fájl keletkezik évente; ha a teljes adat csak néhány GB, akkor egy-egy napi fájl néhány MB-os lehet, és a fájlnyitási, metaadat-olvasási és fájllistázási költség dominál. Ráadásul a nagyon sok apró fájl a schedule- és cache-rétegeket is terheli. Az optimális partíciós szemcsézettség a lekérdezési mintától függ: ha a leggyakoribb szűrés havi, akkor a havi partíció jobb; ha napi, akkor a napi is indokolt lehet, de akkor a partíciók méretét érdemes figyelni.

5. megoldás. A futamok száma:

$$R = \left\lceil \frac{5}{1} \right\rceil = 5 \quad \text{az 1 GB korlátnál},$$

és

$$R = \left\lceil \frac{5}{2} \right\rceil = 3 \quad \text{a 2 GB korlátnál}.$$

A kevesebb futam kevesebb lemez I/O-t és gyorsabb összefésülést jelent, de a nagyobb korlát több RAM-ot foglal el más műveletektől. A spill tehát erőforrás-kompromisszum, nem hiba; a lényeg, hogy a motor ne omoljon össze, ha a munkamemória elfogy.

34.1.7 Gyakori tévedés ellenpéldával

A leggyakoribb tévedés: „A Parquet fájl 2 GB, tehát 2 GB RAM elég lesz.” Ez azért hamis, mert a Parquet tömörített, oszlopos formátum, és a fájlméret nem mond semmit a dekompresszált, objektumfejekkel és indexekkel terhelt memóriareprezentációról. Konkrét ellenpélda: egy fájl 50 millió sort tartalmaz, oszlopai float64, int64 és átlagosan 50 karakteres szöveg. A Parquet tömörített mérete lehet 2 GB. A memóriában a float64 és int64 oszlop együtt $50\,\text{M} \times 16 = 800$ MB, a szöveges oszlop viszont Python-szövegként soronként körülbelül 99 bájtot foglal (49 bájt objektumfej + 50 bájt karakter), ami $50\,\text{M} \times 99 \approx 4.95$ GB. A teljes DataFrame így 5–6 GB RAM-ot igényelhet, miközben a fájl csak 2 GB. A tömörített méret tehát legfeljebb alsó becslésnek használható, de a memóriaigényt mindig a dekompresszált reprezentációból kell számolni.

Ez visszavezet a Szűk keresztmetszet és mérési terv mérési fegyelméhez: a szűk keresztmetszetet mérni kell, nem a fájlméretből kikövetkeztetni. A Feldolgozási gráf és idempotencia pipeline-szemlélete pedig arra emlékeztet, hogy a beolvasási lépésnek ugyanúgy explicit input- és outputhatára van, mint bármely transzformációnak: ha a beolvasás a teljes táblát a memóriába teszi, akkor az a lépés egy rejtett, nagy erőforrásigényű köztes állapotot hoz létre.

Önellenőrzés. Mielőtt elhiszed, hogy egy lekérdezés skálázódik, tedd fel ezeket a kérdéseket: (1) A kód tényleg csak a szükséges oszlopokat kéri (columns, select, SQL-vetítés)? (2) A szűrés a forrásban történik (WHERE, filters, partíciós feltétel), vagy egy már betöltött DataFrame-en? (3) A motor streaming vagy chunk-alapú feldolgozást végez, vagy SELECT * után in-memory táblát hoz létre? (4) A memória-csúcsot mérem, vagy csak a fájlméretet nézem? Ha bármelyikre nemleges a válasz, előbb az egygépes adatfolyamot javítsd, és csak utána fontold meg a klasztert.

34.1.8 Rövid összefoglaló

A nagyobb adat első kérdése nem a klaszter, hanem a beolvasás mérete. A vetítés csak a szükséges oszlopokat olvassa be, a partícióvágás a feltétel szerint kihagyja a fájlok egy részét, a chunk és a streaming pedig darabonként tartja a memóriában az adatot, így a memóriaigény a sorok számától függetlenné válik. A spill biztonsági szelep, a sok apró fájl viszont új szűk keresztmetszetet hoz létre. A tömörített Parquet mérete nem RAM-igény; a memóriát a dekompresszált reprezentációból kell becsülni. A következő alkalmazás a Dask task graph és partíció Dask task graph és partíció: ott a chunkokból és partíciókból építünk lusta végrehajtási gráfot, és megnézzük, hogyan ütemeződik a munka több processzre.