Narcis Miclaus

8 min de citit

PySpark, de la zero Lecția 4 / 60

Arhitectura Spark: driver, executors, cluster manager

Cum ruleaza efectiv un job Spark pe mai multe masini. Driverul, executorii, cluster manager-ul si modelul de resurse care le leaga.

Poți folosi Spark ani la rând fără să te uiți vreodată la diagrama de arhitectură. Cei mai mulți așa fac. Scriu df.groupBy(...).agg(...), dau click pe run și pleacă. Funcționează până când ceva se strică, un job rămâne blocat la 99%, un executor moare, Spark UI arată bare roșii și shuffle reads de gigaocteți, și atunci „nu prea știu ce rulează și unde” devine o problemă serioasă.

Așa că, înainte să atingem o singură linie de PySpark, hai să construim modelul mental a ce se întâmplă atunci când trimiți un job. Până la finalul lecției ăsteia vei putea să te uiți la Spark UI și să-l citești ca pe un roman: ce JVM face ce, unde e paralelismul și de ce chestia numită „shuffle” e răufăcătorul fiecărei povești despre performanță.

Distribuția personajelor

O aplicație Spark care rulează are, la minimum, trei lucruri:

  1. Un driver, un singur proces JVM care rulează codul tău și coordonează totul.
  2. Un cluster manager, chestia care distribuie mașini (sau, mai exact, sloturi pe mașini) aplicației tale.
  3. Unul sau mai mulți executors, procese JVM care fac efectiv munca.

Cam atât. Fără magie. Fără daemon de fundal pe care ai uitat să-l instalezi. Trei componente, trei responsabilități. Odată ce internalizezi asta, Spark încetează să mai fie o cutie neagră.

Hai să le luăm pe rând.

Driverul

Driverul e JVM-ul care ține SparkSession-ul tău. Când rulezi spark-submit my_job.py, sau când pornești un notebook și prima celulă creează o sesiune, driverul e procesul care se lansează. Rulează codul tău Python (sau Scala) de sus în jos, exact așa cum ar face-o un program normal.

Driverul este single-threaded pentru fluxul liniar al codului tău. Buclele for sunt secvențiale. Instrucțiunile if sunt secvențiale. Instrucțiunile print ies în ordine. Driverul e doar un program obișnuit care se întâmplă să știe să vorbească cu un cluster.

Ce face efectiv driverul:

  • Ține SparkSession și SparkContext, punctele tale de intrare în engine.
  • Construiește DAG-ul (Directed Acyclic Graph) de transformări pe măsură ce înlănțui .select(), .filter(), .join(), .groupBy(). Nimic nu se execută încă, driverul doar colectează o rețetă.
  • Când apelezi un action (.count(), .show(), .write.parquet(...), .collect()), driverul predă DAG-ul către Catalyst optimizer, care îl rescrie într-un plan fizic eficient.
  • Apoi driverul împarte planul în stages și tasks și trimite acele tasks către executori.
  • Primește rezultatele înapoi, monitorizează progresul, repornește task-urile eșuate și, în final, returnează răspunsul către codul tău.

Două lucruri de reținut despre driver. Primul: când apelezi .collect(), fiecare rând al rezultatului ajunge înapoi în heap-ul JVM al driverului. Dacă faci .collect() pe un tabel de un miliard de rânduri, driverul tău va face OOM și job-ul moare. Întreabă-te mereu dacă action-ul tău e „de rezumat” (.count(), .show(20)) sau „tot conținutul” (.collect(), .toPandas()). Al doilea: driverul e un single point of failure. Dacă moare, aplicația ta moare. Cluster manager-ele pot reporni executori în mod transparent, dar nu pot reporni un driver în plin zbor.

Executorii

Executorii sunt JVM-urile care macină numerele. Rulează pe worker nodes (mașini din clusterul tău), iar driverul le dă de lucru. Un cluster Spark tipic are între 2 și 2.000 de executori, în funcție de job și de buget.

Fiecare executor are două resurse care contează:

  • Memoria. Folosită pentru caching de DataFrame-uri pe care ai apelat .cache(), pentru a ține shuffle data, pentru a rula task-uri. O configurezi cu spark.executor.memory=4g sau similar.
  • Cores. Numărul de task-uri pe care un executor le poate rula în paralel. Un executor cu 4 cores rulează 4 task-uri simultan. Configurezi cu spark.executor.cores=4.

Înmulțește executorii cu cores și obții paralelismul total. Un cluster cu 10 executori și 4 cores fiecare poate rula 40 de task-uri simultan. Dacă DataFrame-ul tău are 200 de partiții, acele 200 de task-uri vor fi procesate aproximativ în 5 valuri de câte 40.

Executorii trăiesc cât trăiește aplicația. Pornesc când pornește SparkSession, mor când se termină aplicația. (Dynamic allocation poate adăuga și scoate executori în timpul job-ului, dar e o optimizare de care ne ocupăm mult mai târziu.) Un executor care moare în timpul job-ului, OOM, defecțiune de mașină, glitch de rețea, e înlocuit de cluster manager, iar Spark rerulează orice task-uri erau pe el. Asta e celebra fault tolerance: lineage-ul înseamnă că Spark poate recalcula oricând ce a pierdut.

Un punct subtil dar important: în PySpark, fiecare JVM de executor poate să mai pornească unul sau mai multe procese Python worker dacă job-ul tău folosește Python UDF-uri sau operații pe RDD. JVM-ul și Python workerii comunică printr-un socket, iar datele trebuie serializate între ei. Aici PySpark își câștigă reputația de a fi „mai lent decât Scala”, dar doar dacă folosești efectiv Python UDF-uri. Vom intra în detaliu în lecția 6.

Cluster manager-ul

Cluster manager-ul e stratul care deține mașinile și decide cine le folosește. Spark în sine nu rulează mașini, cere altcuiva să-i dea ceva.

Cele patru cluster managers pe care Spark le suportă:

  • Spark Standalone. Manager-ul propriu, livrat cu Spark. Vine cu distribuția Spark, ușor de configurat, ok pentru clustere mici dedicate. Îl vei vedea în tutoriale și medii de laborator.
  • YARN. Resource manager-ul Hadoop. Dacă firma ta are un cluster Hadoop, și un număr surprinzător încă au în 2026, deși scade an de an, Spark on YARN e setarea implicită. YARN a fost deployment-ul de producție dominant ani la rând.
  • Kubernetes. Opțiunea modernă cloud-native. Spark on K8s a maturizat masiv din 2020 încoace și e acum standardul pentru deployment-uri greenfield pe AWS EKS, GCP GKE, Azure AKS. Databricks, EMR și majoritatea platformelor Spark gestionate rulează pe K8s sub capotă. Dacă pornești un nou deployment Spark în 2026, asta e aproape sigur ce vrei.
  • Mesos. Cândva contracandidat, acum aproape mort. Apache a retras Mesos în 2021. Îl vei vedea doar în instalări legacy.

Treaba cluster manager-ului e îngustă: când driverul tău cere „dă-mi 10 executori cu 4 cores și 8 GB de RAM fiecare”, cluster manager-ul găsește mașini cu acele resurse libere și pornește JVM-urile de executor acolo. Odată ce sunt în picioare, driverul vorbește direct cu executorii. Cluster manager-ul e în mare parte în afara imaginii în timpul muncii efective.

De obicei nu alegi tu un cluster manager, platforma alege pentru tine. Databricks alege Kubernetes (varianta lor proprie). EMR te lasă să alegi între YARN și Kubernetes. Glue alege ceva proprietar. De cele mai multe ori scrii același cod PySpark indiferent, iar manager-ul e invizibil.

Cum rulează efectiv un job

Acum hai să urmărim ce se întâmplă când apelezi un action. Să zicem că rulezi:

result = (
    spark.read.parquet("s3://bucket/orders/")
         .filter("order_status = 'shipped'")
         .groupBy("country")
         .agg({"amount": "sum"})
         .collect()
)

Iată secvența:

  1. Primele trei linii (read, filter, groupBy, agg) construiesc un DAG în driver. Nicio dată nu e citită încă.
  2. .collect() e un action. Driverul trimite DAG-ul către Catalyst, query optimizer-ul. Catalyst îl rescrie: împinge filtrul în citirea Parquet, ca să nu citim niciodată rânduri care nu sunt shipped; alege o strategie de agregare; produce un plan fizic.
  3. Planul fizic e împărțit în stages. Un stage e o porțiune contiguă de muncă unde fiecare operație poate fi făcută independent per partiție, fără mișcare de date. Filtrul și agregarea parțială pot avea loc în stage 1. groupBy peste toate datele necesită mișcare de date (un shuffle), deci agregarea finală e în stage 2.
  4. Driverul cere cluster manager-ului executori (dacă nu îi are deja).
  5. Stage 1 se lansează ca N tasks, unde N e numărul de partiții de input. Dacă Parquet-ul are 200 de fișiere, N e probabil 200. Fiecare task e o partiție de muncă, trimisă către un core de pe un executor.
  6. Pe măsură ce task-urile din stage 1 termină, scriu rezultatele parțiale pe disc local pe executor, asta e shuffle write.
  7. Stage 2 se lansează. Fiecare task din stage 2 își citește input-ul din output-ul de shuffle al stage 1 prin rețea, shuffle read, și calculează sumele finale per grup.
  8. Rezultatele finale sunt trimise înapoi la driver, care le asamblează într-o listă Python și le returnează codului tău.

Trei termeni de vocabular din secvența asta sunt cei pe care îi vei vedea în Spark UI toată ziua:

  • Job. Ce declanșează un action. Un singur apel .collect() egal un job.
  • Stage. O felie contiguă din DAG fără shuffle la mijloc. Majoritatea job-urilor au între 2 și 10 stages.
  • Task. O partiție dintr-un stage care rulează pe un core de executor. Unitatea de paralelism. Un job tipic are mii de task-uri.

Momentul „aha” în Spark UI

Spark UI (port implicit 4040 pe driver) este cel mai util instrument de diagnostic din întreg ecosistemul. Odată ce înțelegi arhitectura, UI-ul îți spune totul.

Când deschizi tab-ul Executors, te uiți la JVM-urile care fac munca. Memorie folosită, cores active, task-uri terminate, timp de GC, octeți de shuffle citiți și scriși. Dacă un executor a făcut de 10x mai multă muncă decât ceilalți, ai skew și job-ul tău suferă din cauza asta.

Când deschizi tab-ul Stages, te uiți la feliile DAG-ului. Durata fiecărui stage, numărul de task-uri, dimensiunea input-ului, shuffle read/write. Un stage care durează 10 minute când celelalte au luat 30 de secunde e bottleneck-ul, du-te și uită-te la el.

Când deschizi vizualizarea Tasks într-un stage, te uiți la unitățile individuale la nivel de partiție. Durata minimă, mediană și maximă a task-ului îți spune dacă munca e echilibrată. Dacă maximul e de 50x mediana, ai o strategie proastă de partiționare și câteva rânduri fac munca a milioane.

Asta e bucla completă a optimizării performanței Spark: citește UI-ul, găsește stage-ul lent, găsește task-ul cu skew, repară layout-ul de date. Aproape orice altceva e o notă de subsol.

Ce nu am acoperit încă

Câteva lucruri sărite intenționat, fiindcă sunt lecții întregi prin ele însele:

  • Layout-ul de memorie al unui executor (storage memory, execution memory, unified memory manager). Lecția 14.
  • Internele de shuffle, sort-shuffle, push-based shuffle, rolul external shuffle service. Lecția 22.
  • Adaptive Query Execution (AQE), care rescrie dinamic planul în timpul job-ului bazat pe dimensiunile reale ale partițiilor. Lecția 24.
  • Cluster autoscaling și dynamic allocation, Spark adăugând și scoțând executori pe măsură ce descoperă că are nevoie de ei. Lecția 41.

Pentru moment, ține în minte imaginea: un driver, mulți executors, un cluster manager care distribuie mașinile și un DAG de muncă care curge din SparkSession-ul tău în jos până la partițiile de pe disc. Tot restul în PySpark e detaliu.

Lecția următoare: cele trei API-uri pe care Spark ți le oferă, RDD, DataFrame, Dataset, și de ce pentru ~99% din muncă atingi mereu doar unul dintre ele.

Referințe

Consultat 2026-05-01.

Caută