34.4 Elosztott kísérlet és adatelhelyezés

34.4.1 Cél és konkrét helyzet

Egy csapat a Northstar-rendelésekből havi aggregált feature-táblát épít. A nyers Parquet három skálán létezik: 100 MB szűkített mart, 10 GB rendelési esemény, és 1 TB nyers kattintási napló plusz teljes rendelési előzmény. A kérdés: mikor érdemes Dask/Spark klasztert indítani, és mikor elég a helyi DuckDB vagy Polars? A döntést a számítás, az adat, a hálózat és a hozzáférési jogosultság közös költségmodellje adja, nem a divat. Az alfejezet célja, hogy döntési lapot készíts három adatskálához, konkrét helyi alternatívával és a klaszter indokoltságának feltételével.

Az előzetes eredménybecslés: 100 MB helyi gépen másodpercek, 10 GB percek, 1 TB órák lehet; a klaszter csak akkor győz, ha az adat már a számítás közelében, object store-ban van, és a feldolgozás párhuzamosítható. Ez a becslés nagyságrendi, nem mérés.

34.4.2 Fogalmi magyarázat

Object store. Az adat nem a munkagép helyi fájlrendszerén van, hanem hálózati tárolóban, például S3-kompatibilis object store-ban. Az objektumok statikus fájlok, HTTP-n olvashatók. A Dask/Spark worker feladatai a partíciókat a saját memóriájukba töltik. Ez megosztást és replikációt ad, de hálózati késleltetést és hívási díjat is. Egy 1 TB-os adat 1000 darab 1 GB-os partícióban több ezer hálózati kérés, de a párhuzamos letöltés ezt részben elfedi.

Network. A hálózati átvitel első közelítésben

$$t_{net}=\frac{S}{B},$$

ahol $S$ az adatmennyiség bájtban, $B$ a sávszélesség bájt/másodpercben. Például $S=1$ GB $=10^9$ B, $B=25$ MB/s $=25\cdot 10^6$ B/s esetén $t_{net}=40$ s. Sok kis fájlnál a kérésenkénti késleltetés dominál, ezért a partícióméretet a sávszélességhez kell igazítani. Az elosztott feldolgozásnál a hálózatot kétszer is terheli: az adat beolvasása és a shuffle köztes adat átvitele.

Serialization (sorbarendezés). A driver és az executor/worker közötti objektumokat bájtsorozattá kell alakítani. A Python pickle-t használ, a Spark saját sorosítót. A nagy DataFrame kollekciója a driverre költséges; csak aggregált eredményt gyűjtsünk. A méret becsülhető: $n$ sor $\times$ oszloponkénti bájt. Egy 1 millió soros, 10 oszlopos lebegőpontos tábla 8 bájt/oszlop mellett 80 MB.

Data locality (adatelhelyezés). A számítás akkor gyors, ha a kód az adat mellett fut, nem pedig az adat utazik. A lokális DuckDB az SSD-n lévő adatot olvassa; a klaszter workerje a hálózatról tölti le. Ha az adat már az object store-ban van, és a worker ugyanabban a régióban fut, a párhuzamosság részben kompenzálja a távolságot. A partíciók elhelyezése tehát döntési tényező.

Failure (meghibásodás). A klaszter egy-egy feladata elbukhat: node kiesik, időtúllépés, lemez tele. A Spark lineage újraszámolja az elveszett partíciót; a Dask is újraütemez. Az újraütemezés csak akkor működik, ha a feladat idempotens: azonos bemenet és kimeneti útvonal azonos eredményt ad. A hosszú shuffle-t érdemes checkpointolni, különben egy késői hiba a munka nagy részét újraszámolja.

Reproducibility. A klaszteres futtatás reprodukálásához rögzíteni kell a kód verzióját, a környezetet, a bemeneti adat partícióit és hashét, a seedet, a shuffle partíciók számát és a klaszter konfigurációját. A lebegőpontos összegzés sorrendje változhat a partíciók ütemezésétől, ezért a numerikus ellenőrzés tűréshatáros, nem bitazonos – ez az Optimálás változatlan jelentéssel equivalence check kiterjesztése.

34.4.3 Minimális munkapélda: döntési lap három adatskálához

Az alábbi kód szemléltető paraméterekkel becsüli a helyi és a klaszteres feldolgozási időt. A paraméterek illusztratívak; teljesítményméréshez valódi hardver, adatméret, ismétlés és tényleges futás kell.

# s136_dontesi_lap.py
# Illusztratív költségmodell; az értékek nem mért teljesítményadatok.

scales_mb = {"100 MB": 100e6, "10 GB": 10e9, "1 TB": 1e12}
local_rate = 80e6          # bájt/mp helyi DuckDB/Polars feldolgozás (illusztratív)
upload_rate = 25e6         # bájt/mp hálózati feltöltés (200 Mbps, illusztratív)
cluster_startup = 300      # mp klaszter üzembe helyezése és felszabadítása
cluster_workers = 8
cluster_per_worker_rate = 40e6  # bájt/mp object store-ból olvasó worker (illusztratív)
cluster_rate = cluster_workers * cluster_per_worker_rate

def becsles(size, data_in_object_store=False):
    local_t = size / local_rate
    upload_t = 0.0 if data_in_object_store else size / upload_rate
    cluster_process_t = size / cluster_rate
    cluster_t = cluster_startup + upload_t + cluster_process_t
    return local_t, cluster_t

for name, size in scales_mb.items():
    local_t_upload, cluster_t_upload = becsles(size, data_in_object_store=False)
    local_t_remote, cluster_t_remote = becsles(size, data_in_object_store=True)
    print(f"{name}: ")
    print(f"  adat lokálisan:   helyi {local_t_upload:.0f}s, klaszter {cluster_t_upload:.0f}s")
    print(f"  adat object store: helyi {local_t_remote:.0f}s, klaszter {cluster_t_remote:.0f}s")

Várható kimenet a fenti paraméterekből számítva (nem mérés):

100 MB: 
  adat lokálisan:   helyi 1s, klaszter 305s
  adat object store: helyi 1s, klaszter 300s
10 GB: 
  adat lokálisan:   helyi 125s, klaszter 731s
  adat object store: helyi 125s, klaszter 331s
1 TB: 
  adat lokálisan:   helyi 12500s, klaszter 43125s
  adat object store: helyi 12500s, klaszter 3425s

Kézi ellenőrzés: 10 GB lokálisan $10^{10} / 8\cdot 10^7 = 125$ s; klaszter feltöltéssel $300 + 10^{10}/2{,}5\cdot 10^7 + 10^{10}/3{,}2\cdot 10^8 \approx 300+400+31=731$ s. Az 1 TB feltöltése $4\cdot 10^4$ s, ezért a klaszter csak akkor győz, ha az adat már az object store-ban van.

Ez a döntési lap mutatja, hogy a klaszter indokoltsága erősen függ attól, hol lakik az adat, és hány ismétlés van. A 10 GB-os adat helyi feldolgozása 2 perc, ami elfogadható; a klaszter startupja 5 perc, tehát nem indokolt. Az 1 TB-os adat helyben órákig fut, ezért ha már object store-ban van, a klaszter 57 percet ígér, de ha fel kell tölteni, a feltöltés 11 óra, ami a klasztert nem teszi előnyössé.

34.4.4 A döntési lap egy lehetséges formája

Három adatskálához konkrét helyi alternatívát és klaszterfeltételt rögzítek:

A táblázatot a kód számszerű becslései támasztják alá. A feltétel pontosabban: a klaszter akkor indokolt, ha (1) az adat már a compute-tal azonos régióban lévő object store-ban van; (2) a feldolgozási feladat partíciónként független vagy kevés shuffle-t igényel; (3) a várható klaszterfutás legalább egy nagyságrenddel rövidebb a helyinél; (4) a fejlesztői és üzemeltetési idő rendelkezésre áll. Ha bármelyik hiányzik, a helyi DuckDB/Polars az alapértelmezés.

34.4.5 Eredmény értelmezése

A becslés nem mérés, hanem döntési keret. A helyi DuckDB azért gyors kis adaton, mert nincs hálózati átvitel és indítási overhead. A klaszter előnye a párhuzamosságból jön, de a startup és adatfeltöltés fix költsége a kis és közepes adatnál nagyobb, mint a megtakarítás. Az 1 TB-os esetben a feltöltés nélküli klaszter 3,6-szor gyorsabb, ami már indokolhatja a komplexitást; a feltöltéssel viszont lassabb, mert a hálózati sávszélesség szűk keresztmetszet. Ez a konkrét számpélda is illusztrálja a közös költségmodell elvet: a döntés a teljes láncra vonatkozik, nem egyetlen erőforrásra.

Az elosztott futás költségét az adatmozgatással, inicializálással és klaszterrezsivel együtt kell mérni, nem csak a számítási fázis idejével.

34.4.6 Feladatok

1. Értelmezés. Magyarázd el saját szóval, miért nem elég az adatméret a klaszter melletti döntéshez. Nevezd meg a költségmodell legalább négy elemét, és mondd meg, melyik hiánya okozhat túl korai klaszterválasztást.

2. Implementálás. Módosítsd a fenti kódot úgy, hogy az ismétlésszámot is figyelembe vegye: ha egy elemzést $N$ alkalommal futtatnak, a klaszter startupja amortizálódik. Számítsd ki $N=10$ ismétléssel a 10 GB-os eset döntését, és adj meg egy olyan $N$ küszöböt, ahol a klaszter (adat object store-ban) gyorsabb a helyinél.

3. Hibadiagnózis. Egy Dask-kísérletben a rendeléseket (orders) és a tételeket (order_items) szeretnénk order_id szerint joinolni. Az orders partíciói customer_id szerint, az order_items partíciói line_id szerint vannak; a join 1 óráig fut. Mi a hiba, és milyen javítás csökkenti a shuffle-t? Adj konkrét műveletet.

4. Megváltozott feltétel. Tegyük fel, hogy az 1 TB-os adat már az object store-ban van, de a hálózati olvasási sebesség a worker és a store között 10 MB/s-ra csökken, miközben a helyi gép változatlan. Számold ki az új klaszteres időt, és döntsd el, hogy a klaszter továbbra is indokolt-e. Milyen adatelhelyezési döntést javasolnál?

34.4.7 Megoldások és indoklás

1. Az adatméret önmagában nem mondja meg, hol lakik az adat, mennyi a hálózati sávszélesség, mekkora az indítási overhead, hány ismétlés van, és milyen a feladat párhuzamossága. A költségmodell elemei: adatelhelyezés (adat a gépen vagy object store-ban), hálózati átvitel (feltöltés/letöltés), számítási kapacitás (magok, memória), indítási és üzemeltetési idő, valamint a hibakezelés és reprodukálhatóság többletköltsége. Túl korai klaszterválasztást okoz, ha például az adat feltöltési idejét elfelejtik: 1 TB feltöltése 25 MB/s mellett 11 óra, ami önmagában hosszabb, mint a helyi feldolgozás 3,5 órája.

2. Módosított kód vázlat:

def becsles_n(size, n_runs, data_in_object_store=False):
    local_t_run = size / local_rate
    upload_t = 0.0 if data_in_object_store else size / upload_rate
    cluster_t_run = size / cluster_rate
    return n_runs * local_t_run, cluster_startup + n_runs * (upload_t + cluster_t_run)

# 10 GB, N=10, object store-ban:
print(becsles_n(10e9, 10, data_in_object_store=True))
# Várható: (1250.0, 612.5)  -> klaszter 612s, gyorsabb

10 GB, $N=10$, object store-ban: helyi $10 \cdot 125 = 1250$ s, klaszter $300 + 10 \cdot 31 = 610$ s, tehát a klaszter gyorsabb. A küszöb: $N \cdot 125 > 300 + N \cdot 31$, azaz $N > 300/94 \approx 3{,}2$, tehát 4 vagy több futtatásnál a klaszter (object store-ban) gyorsabb. Az ellenőrzés: $N=4$: helyi 500 s, klaszter 424 s.

3. A hiba a nem egyező particionáló kulcs. A join order_id szerint történik, de az egyik tábla customer_id, a másik line_id szerint particionált. A Dask/Spark emiatt mindkét táblát újraosztja order_id szerint, ami teljes shuffle-t okoz. Javítás: orders = orders.shuffle("order_id") és order_items = order_items.shuffle("order_id"), majd a join; vagy ha az egyik tábla kicsi, broadcast join (dask.dataframe.merge a broadcast=True opcióval). A particionálás utáni kvantitatív ellenőrzés: a shuffle előtti és utáni partíciók számát és a futásidőt összehasonlítani, ügyelve arra, hogy a numerikus eredmény változatlan maradjon (equivalence check tűréshatárral, Optimálás változatlan jelentéssel).

4. Az új klaszteres olvasási idő 10 MB/s mellett: $10^{12} / (8 \cdot 10^7) = 12500$ s (mert 8 worker $\times$ 10 MB/s = 80 MB/s), plusz startup 300 s, összesen 12800 s. Ez gyakorlatilag megegyezik a helyi 12500 s-mal, tehát a klaszter már nem indokolt. Javaslat: az adat és a compute ugyanabban a hálózati zónában legyen; ha ez nem lehetséges, helyi gépre vagy nagyobb sávszélességű kapcsolatra van szükség. Az adatelhelyezés döntése tehát elsődleges a puszta klaszterkapacitás előtt.

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

A leggyakoribb tévedés, hogy a felhő vagy klaszter automatikusan olcsóbb, gyorsabb vagy egyszerűbb. Ellenpélda a 100 MB-os mart táblánál: a helyi DuckDB egyetlen másodperc alatt kiszámol egy csoportosítást, míg a klaszter indítása 5 perc, és a kód átvitele, a függőségek telepítése és a hibakeresés további órákat vehet igénybe. A klaszter itt nem gyorsabb, csak drágább és bonyolultabb. Ugyanez a helyzet a 10 GB-os adatnál, ha az elemzést egyszeri alkalommal futtatják. A klaszter akkor nyer, ha az adat már object store-ban van, a feladat párhuzamosítható, és a futtatás többször ismétlődik. Ezt a feltételt a döntési lap expliciten tartalmazza.

34.4.9 Önellenőrzés

• Sorold fel a költségmodell hat elemét.
• Számold ki: 500 MB adat, 50 MB/s helyi feldolgozás, 300 s startup, 8 worker 25 MB/s sebességgel: érdemes-e klasztert indítani, ha az adat lokális? Válasz: helyi 10 s, klaszter legalább 300 s + feltöltés; nem érdemes.
• Miért nem bitazonos a lebegőpontos összegzés a klaszterben? Mert a partíciók ütemezése változtathatja az összegzési sorrendet, a lebegőpontos összeadás pedig nem asszociatív; ezért relatív tűréshatáros equivalence checket használunk.

34.4.10 Rövid összefoglaló

Az elosztott kísérlet ugyanazon tudományos fegyelmet követi, mint a helyi, de a költségmodell bővül: az object store, a hálózat, a sorbarendezés, az adatelhelyezés, a meghibásodás és a reprodukálhatóság együtt döntik el, hogy a klaszter indokolt-e. A döntési lap három adatskálán mutatja, hogy a helyi DuckDB/Polars kis és közepes adaton gyorsabb, a klaszter pedig akkor nyer, ha az adat már a compute közelében van, a feladat párhuzamos, és az indítási költség amortizálódik. A felhő nem automatikusan jobb; a választást becsült vagy mért teljes láncra kell alapozni.