Ez a blog célja, hogy tisztázzon néhány kezdő problémát, amikor újoncok kódolnak a Spark elosztott számítástechnikához. Az API-k megismerésén kívül fel kell szerelkezni a klaszter részleteivel, hogy a Spark teljesítményét a lehető legjobban kihasználhassuk.

A kiindulási pont a klaszter üzemmód áttekintése lenne .

És néhány gyakori kérdés, ami felmerülhet:

1. Még mindig nem érti a különböző folyamatokat a Spark Standalone klaszterben és a párhuzamosságot.
2. Elfuttatta a bin\start-slave.sh-t, és azt találta, hogy spawnolja a munkást, ami valójában egy JVM. A worker egy JVM folyamat vagy sem?
3. A fenti link szerint a végrehajtó egy olyan folyamat, amelyet egy alkalmazáshoz indítanak egy munkás csomóponton, amely feladatokat futtat. A végrehajtó szintén egy JVM.
4. A végrehajtók alkalmazásonként vannak. Akkor mi a szerepe egy worker-nek? Összehangol a végrehajtóval és az eredményt visszajelzi a vezérlőnek? vagy a vezérlő közvetlenül a végrehajtóval beszél? Ha igen, akkor mi a dolgozó célja?
5. Hogyan lehet szabályozni a végrehajtók számát egy alkalmazáshoz?
6. Meg lehet-e tenni, hogy a feladatok párhuzamosan fussanak a végrehajtón belül? Ha igen, hogyan lehet beállítani a szálak számát egy végrehajtóhoz?
7. Mi a kapcsolat a worker, a végrehajtók és a végrehajtó magok között ( – total-executor-cores)?
8. Mit jelent az, hogy több worker van csomópontonként?

Menjünk vissza a Spark Cluster üzemmód részleteire.

A Spark master/slave architektúrát használ. Amint az ábrán látható, van egy központi koordinátor (Driver), amely sok elosztott munkással (executor) kommunikál. A driver és az egyes végrehajtók saját Java-folyamatokban futnak.

A driver az a folyamat, amelyben a fő metódus fut. Először átalakítja a felhasználói programot feladatokká, majd ütemezi a feladatokat a végrehajtókon.

Spark alkalmazás – -> Driver – -> List of Tasks – -> Scheduler – -> Executors

EXECUTORS

A végrehajtók a munkás csomópontok folyamatai, amelyek egy adott Spark feladat egyes feladatainak végrehajtásáért felelősek. A Spark-alkalmazás elején indítjuk őket, és jellemzően az alkalmazás teljes élettartama alatt futnak. Miután lefuttatták a feladatot, elküldik az eredményeket a meghajtónak. Emellett memórián belüli tárolást biztosítanak az RDD-k számára, amelyeket a felhasználói programok a Block Manager segítségével gyorsítótárba helyeznek.

A végrehajtók indításakor regisztrálják magukat a meghajtóban, és innentől kezdve közvetlenül kommunikálnak egymással. A munkások feladata, hogy közöljék a klasztermenedzserrel az erőforrásaik elérhetőségét.

Egy önálló klaszterben egy végrehajtó jut egy munkásra, kivéve, ha a spark.executor.cores-szal játszunk, és egy munkásnak elég magja van ahhoz, hogy egynél több végrehajtót tartson.

• Egy önálló fürt 5 munkás csomóponttal (minden csomópontnak 8 magja van) Amikor elindítok egy alkalmazást alapértelmezett beállításokkal.
• A Spark mohón annyi magot és végrehajtót fog szerezni, amennyit az ütemező felajánl. Így a végén 5 végrehajtó lesz egyenként 8 maggal.

A következők a spark-submit beállításai a végrehajtók számával való játékhoz:

– executor-memory MEM Memória végrehajtónként (pl. 1000M, 2G) (Alapértelmezett: 1G).

Spark standalone és YARN csak:
– executor-cores NUM Magok száma végrehajtónként. (Alapértelmezett: 1 YARN módban, vagy az összes rendelkezésre álló processzormag a dolgozóban standalone módban)

Kizárólag YARN:
– num-executors NUM Az indítandó végrehajtók száma (Alapértelmezett: 2). Ha a dinamikus allokáció engedélyezve van, a végrehajtók kezdeti száma legalább NUM lesz.

• A 2 worker instance egy worker node-ot jelent 2 worker processzel?

A node egy gép, és nincs jó ok arra, hogy gépenként egynél több worker-t futtassunk. Tehát két worker node tipikusan két gépet jelent, mindegyik egy Spark worker.

1 Node = 1 Worker process

• Minden worker instance egy végrehajtót tart egy adott alkalmazáshoz (ami tárolást, feladatot kezel), vagy egy worker node egy végrehajtót tart?

A workerek sok végrehajtót tartanak, sok alkalmazáshoz. Egy alkalmazásnak sok munkáson vannak végrehajtói

Egy munkás csomópont több végrehajtót (folyamatot) is tarthat, ha elegendő CPU-val, memóriával és tárolóval rendelkezik.

• BTW, a végrehajtók száma egy munkás csomópontban egy adott időpontban teljes mértékben a fürtön lévő munkaterheléstől és a csomópont képességétől függ, hogy hány végrehajtót futtasson.

ALkalmazás végrehajtási folyamata

Ezt szem előtt tartva, amikor a spark-submit segítségével elküldünk egy alkalmazást a fürtbe, belsőleg a következő történik:

1. Egy önálló alkalmazás elindul és instanciál egy SparkContext/SparkSession példányt (és csak ezután lehet az alkalmazást meghajtónak hívni).
2. A driverprogram erőforrásokat kér a fürtkezelőtől a végrehajtók indításához.
3. A fürtkezelő elindítja a végrehajtókat.
4. A driverfolyamat a felhasználói alkalmazáson keresztül fut. Az RDD-ken végzett műveletektől és transzformációktól függően feladatokat küld a végrehajtóknak.
5. A végrehajtók lefuttatják a feladatokat és elmentik az eredményeket.
6. Ha valamelyik munkás összeomlik, a feladatait más végrehajtóknak küldi el, hogy újra feldolgozzák. A Learning Spark című könyvben: Lightning-Fast Big Data Analysis” című könyvben beszélnek a Sparkról és a hibatűrésről:

A Spark automatikusan kezeli a hibás vagy lassú gépeket a hibás vagy lassú feladatok újrafuttatásával. Ha például egy map() művelet partícióját futtató csomópont összeomlik, a Spark újra lefuttatja azt egy másik csomóponton; és még ha a csomópont nem is omlik össze, hanem egyszerűen csak sokkal lassabb, mint a többi csomópont, a Spark képes preemptív módon elindítani a feladat “spekulatív” másolatát egy másik csomóponton, és átvenni annak eredményét, ha az befejeződik.

1. A meghajtó SparkContext.stop() parancsával, vagy ha a main metódus kilép/összeomlik, az összes végrehajtó befejeződik, és a klaszter erőforrásokat a klaszterkezelő felszabadítja.

Ha megnézzük a végrehajtást a Spark perspektívából bármely erőforrás-kezelőn keresztül egy olyan programra, amely két rdds-t egyesít és valamilyen redukciós műveletet végez, akkor filter

Az alábbi lista néhány ajánlást rögzít, amelyeket szem előtt kell tartani a konfigurálás során:

• Hadoop/Yarn/OS démonok: Amikor Spark alkalmazást futtatunk egy olyan fürtkezelővel, mint a Yarn, akkor számos daemon fog futni a háttérben, mint a NameNode, Secondary NameNode, DataNode, JobTracker és TaskTracker. Ezért a num-executors megadásakor meg kell győződnünk arról, hogy elegendő magot hagyunk félre (~1 mag csomópontonként), hogy ezek a daemonok zökkenőmentesen fussanak.
• Yarn ApplicationMaster (AM): Az ApplicationMaster felelős az erőforrások megtárgyalásáért a ResourceManagertől, és a NodeManagerekkel együttműködve végrehajtja és felügyeli a konténereket és azok erőforrás-fogyasztását. Ha sparkot futtatunk yarnon, akkor be kell terveznünk az AM-nek szükséges erőforrásokat (~1024MB és 1 Executor).
• HDFS Throughput: A HDFS kliensnek gondjai vannak a rengeteg párhuzamos szálakkal. Megfigyelték, hogy a HDFS teljes írási átbocsátóképességet ~5 feladat végrehajtónként . Tehát jó, ha a végrehajtónkénti magok számát ez alatt a szám alatt tartjuk.
• MemoryOverhead: A következő kép a spark-yarn-memóriahasználatot ábrázolja.

Két dolgot érdemes megjegyezni ebből a képből:

Full memory requested to yarn per executor =
spark-executor-memory + spark.yarn.executor.memoryOverhead.
spark.yarn.executor.memoryOverhead = 
Max(384MB, 7% of spark.executor-memory)

Azaz, ha végrehajtónként 20GB-ot kérünk, AM valójában 20GB + memoryOverhead = 20 + a 20GB 7%-a = ~23GB memória jut nekünk.

• A végrehajtók túl sok memóriával történő futtatása gyakran túlzott szemétgyűjtési késedelmeket eredményez.
• A pici végrehajtók futtatása (például egyetlen maggal és csak annyi memóriával, amennyi egyetlen feladat futtatásához szükséges) elveti azokat az előnyöket, amelyek a több feladat egyetlen JVM-ben történő futtatásából származnak.

Elég az elméletből… Menjünk a gyakorlatba…

Most tekintsünk egy 10 csomópontos fürtöt a következő konfigurációval, és elemezzük a végrehajtók-mag-memória elosztás különböző lehetőségeit:

**Cluster Config:**
10 Nodes
16 cores per Node
64GB RAM per Node

Első megközelítés: Apró végrehajtók :

Az apró végrehajtók lényegében egy végrehajtót jelentenek magonként. A következő táblázat a spar-config paramétereink értékeit mutatja be ezzel a megközelítéssel:

- `--num-executors` = `In this approach, we'll assign one executor per core`
= `total-cores-in-cluster`
= `num-cores-per-node * total-nodes-in-cluster` 
= 16 x 10 = 160
- `--executor-cores` = 1 (one executor per core)
- `--executor-memory` = `amount of memory per executor`
= `mem-per-node/num-executors-per-node`
= 64GB/16 = 4GB

Analízis: Ha magonként csak egy végrehajtó van, ahogy fentebb tárgyaltuk, nem fogjuk tudni kihasználni a több feladat ugyanazon a JVM-en történő futtatásának előnyeit. Továbbá a megosztott/kecsetelt változók, mint például a broadcast változók és az akkumulátorok a csomópontok minden egyes magjában replikálódnak, ami 16-szoros. Továbbá nem hagyunk elég memória overheadet a Hadoop/Yarn daemon folyamatok számára, és nem számolunk az ApplicationManagerben. NEM JÓ!

Második megközelítés:

A kövér végrehajtók lényegében egy végrehajtót jelentenek csomópontonként. Az alábbi táblázat a spark-config paramétereink értékeit mutatja be ezzel a megközelítéssel:

- `--num-executors` = `In this approach, we'll assign one executor per node`
= `total-nodes-in-cluster`
= 10
- `--executor-cores` = `one executor per node means all the cores of the node are assigned to one executor`
= `total-cores-in-a-node`
= 16
- `--executor-memory` = `amount of memory per executor`
= `mem-per-node/num-executors-per-node`
= 64GB/1 = 64GB

Analízis: Végrehajtónként mind a 16 maggal, az ApplicationManager és a daemon folyamatokon kívül nem számolunk, a HDFS áteresztőképessége sérülni fog, és ez túlzott szemét eredményeket fog eredményezni. Továbbá,NEM JÓ!

Harmadik megközelítés: Egyensúly a Fat (vs.) Tiny között

A fentebb tárgyalt ajánlásoknak megfelelően:

• A fent említett ajánlások alapján rendeljünk 5 magot végrehajtóként => – végrehajtó-magok = 5 (jó HDFS áteresztőképességhez)
• Hagyjunk 1 magot csomópontonként a Hadoop/Yarn démonok számára => Node-onként rendelkezésre álló magok száma = 16-1 = 15
• So, A klaszterben összesen rendelkezésre álló magok száma = 15 x 10 = 150
• A rendelkezésre álló végrehajtók száma = (összes mag / magok száma végrehajtónként) = 150 / 5 = 30
• 1 végrehajtót hagyunk az ApplicationManager számára => – num-végrehajtók = 29
• Futtatók száma csomópontonként = 30/10 = 3
• Memória végrehajtónként = 64GB / 3 = 21GB
• Leszámítva a halom overheadet = 21GB 7%-a = 3GB. Tehát tényleges – végrehajtó-memória = 21-3 = 18GB

Az ajánlott konfiguráció tehát: 29 végrehajtó, egyenként 18GB memória és egyenként 5 mag!!!

Analízis: Nyilvánvaló, hogy ez a harmadik megközelítés megtalálta a megfelelő egyensúlyt a Fat vs. Tiny megközelítések között. Mondanom sem kell, hogy egy kövér végrehajtó párhuzamosságát és egy apró végrehajtó legjobb teljesítményét érte el!!!

Következtetés:

Láttuk:

• Pár ajánlást, amit szem előtt kell tartani, amikor ezeket a paramétereket konfiguráljuk egy Spark-alkalmazáshoz, mint például:
• Bevetni az erőforrásokat, amelyekre a Yarn alkalmazáskezelőjének szüksége lenne
• Hogyan tartalékoljunk néhány magot a Hadoop/Yarn/OS deamon folyamatok számára
• Megtudtuk a spark-yarn-memory-usage
• Megnéztünk és elemeztünk továbbá három különböző megközelítést ezen paramok konfigurálására:

1. Tiny Executors – One Executor per Core
2. Fat Executors – One executor per Node
3. Recepted approach – Right balance between Tiny (Vs) Fat coupled with the recommendations.

– num-executors, – executor-cores és – executor-memory.. ez a három paraméter nagyon fontos szerepet játszik a spark teljesítményében, mivel ezek szabályozzák a CPU & memória mennyiségét, amit a spark alkalmazás kap. Ez teszi nagyon fontossá, hogy a felhasználók megértsék a helyes konfigurálásuk módját. Remélem, ez a blog segített neked abban, hogy megkapd ezt a perspektívát…