34.3 Spark DataFrame és végrehajtási terv

34.3.1 Cél és konkrét helyzet

A Dask task graph és partíció Dask-példában már láttunk lusta feladatgráfot és partíciók közötti műveletet. Itt a Spark ugyanazt a kérdést más futásidejű szerződéssel oldja meg: egy vezérlő folyamat felépíti a végrehajtási tervet, a végrehajtók pedig partíciókonként dolgoznak. A tanulási helyzet nem valódi klaszter, hanem helyi, CPU-s mód, a cél az aggregálás tervezése és korrekt, kis előnézetű futtatása.

Bemenet: a Northstar lab/orders.csv, 2400 rendelés, minden sor egy megfigyelési egység, order_id kulccsal. A használt oszlopok: priority_flag (0/1, expressz jelzés), basket_minor (egész EUR-cent kosárérték), quantity_sum, distance_km. Cél: priority_flag szerint rendelésszám, kosárösszeg és átlagkosár. Elvárt ellenőrzés: a kimenet két csoportot tartalmaz, a két basket_minor-összeg összege a teljes fájl kosárösszegét adja, és a rendelésszámok összege 2400. Előzetes becslés: a kimenet 2 sor, 4 oszloppal; a futás helyi módban a 2400 soros fájlon gyors, de ennek mért idejét itt nem állítjuk.

34.3.2 Fogalmi magyarázat: driver, executor, terv, shuffle

A Spark-folyamatnak két fő szereplője van:

• driver: a vezérlő folyamat. Lefuttatja a programodat, minden lusta transzformációból felépíti a tervet, ütemezi a munkát, koordinálja a végrehajtókat, és az akciók kis eredményét visszagyűjti.
• executor: a végrehajtó folyamat. Partíciókat tárol, oszloparányosan olvas, számol, és a shuffle során adatot küld más végrehajtóknak.

A Spark DataFrame elosztott, oszlopos, lusta reláció. Ugyanúgy van sémája, sora és oszlopa, mint egy pandas vagy Polars táblának, de a sorok partíciókra osztva élnek az executorokon. Transzformáció (transformation) minden olyan művelet, amely új DataFrame-et ad vissza, de nem indít azonnal számítást: select, filter, withColumn, groupBy, agg, join, orderBy stb. Akció (action) az, ami tényleges számítást indít, és eredményt hoz vissza a driverre vagy külső tárolóba ír: count, collect, show, take, write.

Az explain ezért nem futtató parancs: kiíratja a strukturált tervet, hogy az áthelyezések előtt átgondolhasd az adatmozgást. Például:

HashAggregate(keys=[priority_flag], functions=[finalmerge_sum, finalmerge_count])
+- Exchange hashpartitioning(priority_flag, 200)
   +- HashAggregate(keys=[priority_flag], functions=[partial_sum, partial_count])
      +- FileScan csv [...] [priority_flag,basket_minor]

A tervet alulról felfelé olvasom: először a CSV beolvasása, utána partíción belüli részleges összegzés, majd shuffle, végül a kulcs szerinti végleges összegzés.

A partition a párhuzamosság fizikai egysége: az adat egy darabja, amelyet egy executor-mag dolgoz fel. A shuffle partíciók közötti adatáthelyezés kulcs szerint. Az aggregálásnál a Spark hash particionálást használ: minden (priority_flag) kulcs a hash alapján kiválasztott partícióba kerül, hogy minden csoport egyetlen helyen legyen. A shuffle drága, mert lemezre és hálózatra is kerülhet adat, ezért a Spark először lokális részösszegeket számít.

A collect limit a biztonságos előnézet mintája: a .limit(2).show() vagy .take(2) azt mondja, hogy a driverre legfeljebb két sor érkezzen. A teljes nagy táblát .collect()-elni tilos, mert a driver memóriájába gyűjtöd a teljes elosztott adatot, ami tipikus OOM-hiba.

A relációs aggregáció matematikája ugyanaz, mint Reláció, NULL és elemzési szemcsézettség alfejezetben. Legyen $P_k$ a $k$-adik partíció sorhalmaza, $x_i$ a basket_minor érték az $i$. rendelésben. Az aggregálás előtti részösszeg:

$$S_k = \sum_{i \in P_k} x_i, \qquad N_k = \sum_{i \in P_k} 1,$$

ahol $S_k$ a partíció kosárösszege EUR-centben, $N_k$ a rendelésszám a partíción. A shuffle kulcs szerint összegyűjti a részösszegeket, a végleges csoportösszeg:

$$S_g = \sum_{k} S_{k,g}, \qquad N_g = \sum_{k} N_{k,g},$$

ahol $g$ a csoportkulcs (0 vagy 1), $S_{k,g}$ és $N_{k,g}$ a $k$ partícióban a $g$ csoporthoz tartozó részösszeg. Az átlag csak ezután osztás:

$$\bar{x}_g = \frac{S_g}{N_g} \quad \text{EUR-cent/rendelés, ha } N_g > 0.$$

Kézi kontroll három soron: [(0, 500), (1, 700), (0, 300)]. Ekkor két partícióra bontva például az elsőben (0, 500), (1, 700), a másodikban (0, 300): az egyes csoportra $S_{0,1}=500$, $N_{0,1}=1$, $S_{0,2}=300$, $N_{0,2}=1$, így $S_0=800$, $N_0=2$, $\bar{x}_0=400$; az egyes csoportra $S_1=700$, $\bar{x}_1=700$. Az invariáns $S_0 + S_1 = 1500$, ami a teljes összeg.

34.3.3 Minimális munkapélda

Platform: Linux, munkakönyvtár ~/northstar-lab, külön bővítő Python-környezet, mert PySpark nincs a kezdő lab/pyproject.toml alapfüggőségei között. A pontos telepített verziót a pip show pyspark paranccsal ellenőrizd; az alábbi kódrészlet a PySpark 3.5.x API-ját szemlélteti. Ennek a kódrészletnek a futtatását e könyvfejezet generálásakor nem végeztük el: ahol kimenetet írunk, az elméletileg várt vagy a tervből következő szemléltetés, nem mért futási jegyzék.

# Linux, ~/northstar-lab
python -m venv .spark-venv
source .spark-venv/bin/activate
python -m pip install pyspark  # a pontos verzió rögzítését a projekt lockjában végezd el

from pyspark.sql import SparkSession, functions as F

spark = (
    SparkSession.builder
    .appName("s135-region-aggregation")
    .master("local[2]")          # helyi, 2 szálú mód csak tanulásra
    .getOrCreate()
)
spark.sparkContext.setLogLevel("WARN")

# Explicit séma: a Northstar orders.csv teljes adatszerződése.
orders = spark.read.schema("""
    customer_id string,
    order_id string,
    order_ts string,
    promised_days int,
    quantity_sum int,
    item_rows int,
    basket_minor long,
    shipping_fee_minor long,
    currency string,
    distance_km double,
    priority_flag int,
    delivery_days int,
    actual_delivery_at string,
    label_available_at string,
    target_late int
""").option("header", True).csv("lab/orders.csv")

# Lusta transzformációk: semmi nem számolódik itt.
grouped = (
    orders
    .groupBy("priority_flag")
    .agg(
        F.count("*").alias("order_count"),
        F.sum("basket_minor").alias("basket_sum_minor"),
        F.avg("basket_minor").alias("avg_basket_minor"),
    )
)

# Nem futtat, csak kiíratja a fizikai tervet.
grouped.explain("formatted")

# Biztonságos előnézet: a driverre legfeljebb két sor kerül.
preview = grouped.orderBy("priority_flag").limit(2)
preview.show()

# Helytelen minta nagy adatra; itt 2400 sorral még elfutna, de a szokást ne építsd be:
# all_rows = orders.collect()

spark.stop()

A .groupBy().agg() három új oszlopot ad: order_count hosszú egész, basket_sum_minor hosszú egész, avg_basket_minor dupla lebegőpontos. A grouped.explain("formatted") nem indít munkát; a tényleges számítást a preview.show() akciója indítja el. A local[2] mester két executor-szálat enged ugyanabban a JVM-ben, ezért a shuffle nem valódi hálózati út, hanem folyamaton belüli adatáthelyezés.

34.3.4 Eredmény értelmezése

A várt kimeneti séma: priority_flag int, order_count long, basket_sum_minor long, avg_basket_minor double; sorok száma legfeljebb kettő, mert ennyi különböző csoport van. A show() nem garantálja a rendezést, ezért a determinista előnézethez orderBy("priority_flag")-et adtunk.

A fizikai terv elvárt kulcsai: FileScan csv a beolvasáshoz; részleges HashAggregate csoportonként; Exchange hashpartitioning(priority_flag) a shuffle-höz; végleges HashAggregate a csoportösszegek összefésüléséhez. A limit(2) a tervben CollectLimit vagy TakeOrderedAndProject jellegű lépésként jelenhet meg, és garantálja, hogy a driverre csak két sor érkezik.

Ellenőrző invariánsok:

1. $\sum_g \text{order\_count}_g = 2400$, mert minden rendelés pontosan egy csoportban van.
2. $\sum_g \text{basket\_sum\_minor}_g = \sum_i \text{basket\_minor}_i$, egész centben pontosan, ennek minden adatsorban nullának kell lennie.
3. A súlyozott átlag az egész táblára: $$\bar{x}_{\text{össz}} = \frac{\sum_g N_g \bar{x}_g}{\sum_g N_g},$$ ahol $N_g$ a order_count, $\bar{x}_g$ az avg_basket_minor. Lebegőpontos osztás miatt a két oldal közötti eltérésnek $10^{-6}$ cent/rendelés relatív tűrésen belül kell lennie.

Mivel az avg_basket_minor lebegőpontos, az összegző ellenőrzést ne abból, hanem a basket_sum_minor-ból számold: a sum * count szorzat összege nem feltétlenül adja vissza az egész összeget kerekítési hiba nélkül. A pontos, egész centes ellenőrzés mindig a basket_sum_minor oszlopból származzon.

A Spark explain logikai és fizikai tervet mutat; a tényleges futási terv és a shuffle mennyisége eltérhet az előzetes becsléstől.

34.3.5 Feladatok

1. Tervértelmezés. A fenti grouped.explain("formatted") kimenetében azonosítsd a részleges aggregálást, a shuffle-t és a végleges aggregálást! Melyik lépés igényel partíciók közötti kulcs szerinti adatáthelyezést, és miért van szükség rá?

2. Megvalósítás. Írj PySpark local feladatot, amely a priority_flag szerint számol quantity_sum-összeget és distance_km-átlagot a lab/orders.csv-ből, és take(2)-vel jeleníti meg az eredményt! Ne olvasd be a teljes táblát a driverre! Add meg a várt shape-et és dtype-ot!

3. Hibadiagnózis. Egy kolléga nagy, több millió soros táblán ezt futtatja:

   result = orders.groupBy("priority_flag").count().collect()

   Mi a hiba? Miért nem jelent megoldást, ha az adat kicsi, ezért nálad még lefutott? Adj biztonságos előnézetet!

4. Megváltozott feltétel. A orders.csv-t 32 Parquet-fájlra bontották object storage-on. Hogyan változik a beolvasási és aggregálási terv? Mely részek olcsóbbak vagy drágábbak? Miért nem elég a helyi gépen mért futásidő a klaszteres várható idő becsléséhez?

5. Erőforrásbecslés. A Párhuzamos modellkeresés erőforráskerettel alfejezetben tanult szorzatszabályhoz hasonlóan gondold át: ha egy aggregálás 10 millió soron, 4 kulcsoszlop, összesen 1000 csoport mellett 200 partícióra shuffle-öl, milyen lépésekben nőhet a költség a 2400 soros helyi próbához képest, és miért nem arányos pusztán a sorok számával?

34.3.6 Megoldások és indoklás

1. megoldás. A FileScan beolvassa a CSV-partíciókat. A részleges HashAggregate minden partíción külön-külön számítja a csoportonkénti részösszeget és darabszámot. Az Exchange hashpartitioning(priority_flag) a shuffle: minden részsor a priority_flag hash-alapú partíciójába kerül, hogy az azonos kulcsú részösszegek ugyanazon a helyen találkozzanak. A végleges HashAggregate a részösszegekből csoportösszeget és átlagot készít. A shuffle azért kell, mert egy csoport kulcsa több bemeneti partícióban is előfordulhat.

2. megoldás. Egy lehetséges változat:

orders = spark.read.schema("""
    customer_id string, order_id string, order_ts string,
    promised_days int, quantity_sum int, item_rows int,
    basket_minor long, shipping_fee_minor long, currency string,
    distance_km double, priority_flag int, delivery_days int,
    actual_delivery_at string, label_available_at string, target_late int
""").option("header", True).csv("lab/orders.csv")

summary = (
    orders
    .groupBy("priority_flag")
    .agg(
        F.sum("quantity_sum").alias("total_quantity"),
        F.avg("distance_km").alias("avg_distance_km"),
    )
)

print(summary.take(2))  # legfeljebb két sor; az egész tábla nem kerül a driverre
spark.stop()

Várt shape: (2, 3), oszlopok: priority_flag int, total_quantity long, avg_distance_km double. Ellenőrzés: a két total_quantity összege a teljes quantity_sum összege; a avg_distance_km mértékegysége km/rendelés.

3. megoldás. A .collect() a teljes csoportosított eredményt a driver memóriájába teszi. Kis adaton ez eltűnik, ezért a hiba nem jelentkezik, de a művelet szemantikája nagy adatra is ugyanaz: az elosztott eredményt egyetlen gép memóriájába gyűjti. A kolléga általában csak meg akarja nézni a táblát, tehát result.limit(5).show() vagy take(5) a helyes. A count()-ból amúgy is csak két sor jönne, tehát ebben a konkrét példában kisebb a veszély, mint az egész orders táblára hívott collect()-nél; a gondolatmenet akkor is ugyanaz: előbb a művelet kimenetének méretét, aztán a driverre kerülő sorok számát kell becsülni.

4. megoldás. Parquet-fájloknál a beolvasás oszloposan és predikátum-leszúrással hatékonyabb lehet, a CSV-nél nem kell sémát fejteni. A 32 fájl akár 32 vagy több bemeneti partícióra bomlik, ezért a részleges aggregálás párhuzamosabb. A shuffle viszont akkor is adatmozgást jelent, ha a fájlok object storage-on vannak: a kulcsok szerinti áthelyezést a hálózat és az executorok közötti kapcsolat korlátozza. A helyi gépen minden egy JVM-ben fut, ezért nincs hálózati késés, nincs storage API-hívás, és a memória is más. A klaszteres időt ugyanazon hardveren, méreten, partíciószámon, ismétléssel kell mérni; a helyi próba csak a terv helyességét ellenőrzi.

5. megoldás. A költség a beolvasott bájtokban, a csoportkulcsok szerializációjában, a shuffle partíciók számában, a hálózatban és a végleges aggregáció memóriaigényében nőhet. A 10 millió soros esetben a shuffle-re kerülő részösszegek mérete a csoportok és partíciók szorzatával nőhet, nem lineárisan a bemenettel. A 200 partíció túl kevés lehet, ha az executorok száma nagy, túl sok, ha kicsi; a becsléshez a Párhuzamos modellkeresés erőforráskerettel erőforrás-szorzóját és a Távolság, kernel és méretkorlát méretkorlát-gondolkodását kell alkalmazni.

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

A leggyakoribb tévedés: „A local módban mért 2,1 másodperc, tehát a klaszteren is körülbelül ennyi lesz.” A local mód egyetlen JVM-ben futtatja a drivert és az executorokat; a shuffle nem hálózati, hanem folyamaton belüli másolás, az object storage-hívások helyett helyi fájl van, és a partíciók nem mennek át sorosításon ugyanúgy. Ezért a kis helyi próba csak a terv szemantikáját és az eredmény invariánsait ellenőrzi, nem a klaszteres teljesítményt. A Dask task graph és partíció Dask-tapasztalat is ugyanezt mutatta: a scheduler és a partíciók közötti művelet költsége csak a tényleges végrehajtó környezetben értelmezhető. A korrekt méréshez gép, adatméret, partíciószám, ismétlésszám és hibahatár kell, ahogyan a Távolság, kernel és méretkorlát alfejezetben a futásidő-állításoknál már rögzítettük. Önellenőrzés: ha a Spark-feladat csak .show()-t hív, akkor még nem mérted a teljes aggregációt; ha .collect()-et hívsz, akkor a drivert méred, nem a klasztert.

34.3.8 Rövid összefoglaló

A Spark DataFrame lusta, oszlopos, partíciókra osztott reláció. A transzformációk tervet építenek, az akciók indítanak számítást. Az explain a tervet mutatja, a groupBy().agg() részleges aggregálást, kulcs szerinti shuffle-t és végleges aggregálást hoz létre. A részösszeg-képlet invariánsa: a csoportösszegek összege a teljes összeg. Nagy adatot nem gyűjtünk a driverre; take, limit, show a biztonságos előnézet. A local módos próba a helyességet ellenőrizheti, de nem klaszteres teljesítménymérés. A következő alkalmazás Elosztott kísérlet és adatelhelyezés: elosztott kísérletek futtatása és az adatelhelyezés tervezése, ahol ezt a tervszemléletet kell adatparticionálással összekötni.