Ce se întâmplă, de fapt, când rulezi un job Spark?

Introducere

Pe măsură ce seturile de date cresc, procesarea lor eficientă pe o singură mașină poate deveni dificilă din cauza limitărilor capacității procesorului, memoriei și debitului de stocare.

Apache Spark răspunde acestor provocări oferind un motor distribuit de procesare a datelor, care poate împărți sarcinile în partiții și poate executa sarcini în paralel pe mai multe nuclee ale procesorului și pe mai multe mașini.

Totuși, Spark nu înseamnă doar adăugarea unui număr mai mare de mașini.

Pentru a proiecta fluxuri de date eficiente și fiabile, este esențial să înțelegem cum interacționează componentele sale — drivere, executori, RDD-uri, DataFrame-uri, partiții, joburi, etape și sarcini.

În acest articol, voi prezenta arhitectura Spark, mecanismele de execuție și principalele aspecte care influențează performanța, folosind un exemplu practic de agregare a vânzărilor din comerțul electronic.

1. De ce Apache Spark – procesarea paralelă a datelor

Apache Spark este un motor de calcul distribuit, conceput pentru procesarea datelor la scară largă și ingineria datelor.

Spark procesează volume mari de date folosind procesarea paralelă a datelor:

Figura A. Arhitectura la nivel înalt a unui cluster distribuit Apache Spark (cluster, driver și noduri de lucru), comparată cu execuția pe un singur nod (creată de Kuriko IWAI)

Spre deosebire de procesarea pe un singur nod (în dreapta, Figura A), Spark împarte activitatea în sarcini folosind mai multe noduri și le execută pe resursele disponibile, pe mai multe mașini.

Aici, termenul „distribuit” nu înseamnă că sunt obligatorii mai multe mașini. Spark poate rula în modul local pe o singură mașină. Chiar și în acest caz, datele pot fi împărțite în partiții și procesate în paralel.

2. Arhitectura de bază a Apache Spark

Spark este alcătuit din mai multe componente-cheie:

• Driverul Spark (driver): procesul central care coordonează aplicația Spark.
• Managerul de cluster: un serviciu extern responsabil de alocarea resurselor hardware, precum nucleele procesorului și memoria.
• Noduri de lucru (worker): mașini fizice sau virtuale ori noduri de calcul din cluster.
• Executori: procese de lucru distribuite, lansate pe nodurile de lucru.
• Java Virtual Machine (JVM): mediul de execuție necesar pentru a rula codul de octeți Spark (deoarece Spark este scris în Scala/Java).

Mai întâi, driverul coordonează o aplicație Spark și o transformă într-un graf orientat aciclic (DAG) de sarcini de execuție fizice, apoi le programează pe executori.

Apoi, managerul de cluster alocă resurse în cadrul sistemului de gestionare a resurselor folosit pentru implementare.

Nodurile de lucru furnizează resurse de calcul pentru executori, care execută sarcini și păstrează datele pe durata executării acestora.

Unele medii gestionate, precum Databricks, folosesc un executor pentru fiecare nod de lucru, dar aceasta nu este o regulă universală în Spark. În funcție de configurația clusterului, un nod de lucru poate găzdui mai mulți executori.

2-1. Caz de utilizare în comerțul electronic – structura datelor de intrare

Să ne imaginăm o companie boutique de comerț electronic.

Aceasta are 10,000 de înregistrări de vânzări, fiecare cu o unitate de stocare (SKU) și o cantitate vândută. Să presupunem că există trei SKU-uri posibile: A, B, și C.

Tabelul 1. Exemplu de defalcare a înregistrărilor de vânzări din comerțul electronic și de agregări estimate.

Scopul nostru este să calculăm numărul total de unități valide vândute pentru fiecare SKU:

1. Eliminați înregistrările ale căror qty <= 0, apoi
2. Agregați eșantioanele după SKU.

În setul de date complet, folosit aici doar cu titlu ilustrativ, să presupunem că după filtrare rămân 9,800 dintre cele 10,000 de înregistrări. Totalurile finale pe care le așteptăm de la Spark sunt:

[("A", 8000), ("B", 10500), ("C", 9200)]

2-2. Clusterul Spark

Voi defini un cluster simplificat pentru cazul nostru de utilizare.

Noduri de lucru (workeri)

Voi configura 3 noduri de lucru, fiecare cu o instanță de 8 vCPU și 32 GiB de RAM:

Figura B. Alocarea resurselor între nodurile de lucru (creată de Kuriko IWAI)

În total, în sistem sunt disponibile 24 de vCPU-uri.

Executori Spark și alocarea CPU/memoriei

Să presupunem că rezerv:

• 7 executori.
• 2 nuclee vCPU pentru fiecare executor.
• 3 GiB de memorie heap JVM pentru fiecare executor.

Aplicația Spark solicită managerului de cluster să aloce resurse pentru acești 7 executori.

Apoi, managerul de cluster decide cum să aloce procesoarele și memoria, în funcție de utilizarea resurselor de către fiecare worker.

De exemplu, decide să creeze 14 sloturi pentru sarcini — 2 nuclee vCPU pentru fiecare executor × 7 executori:

Figura C-1. Nuclee vCPU, memorie și alocarea executorilor între workeri, cu dezechilibru al resurselor în Worker A

Rețineți că cei 3 GiB de memorie alocați executorului nu reprezintă memoria sa totală, deoarece trebuie rezervată memorie suplimentară pentru workeri și alte execuții.

Presupunând că o singură sarcină necesită 1 vCPU (spark.task.cpus = 1), fiecare worker își alocă vCPU-urile:

Figura C-2. Alocarea paralelă a vCPU-urilor între workeri

Acest lucru evidențiază două aspecte esențiale:

1. Fiecare worker are o rată proprie de utilizare a CPU-ului (în Figura C-2, utilizarea în Worker A este de 75%, în timp ce în Workers B și C rămâne la 50%).
2. Deși sistemul are 24 de nuclee vCPU (8 vCPU/worker × 3 workeri), poate folosi doar 14 nuclee vCPU; celelalte 10 rămân neutilizate pentru aceste sarcini.

2-3. Mașina virtuală Java (JVM) din spatele Spark

Apache Spark este construit pe Mașina virtuală Java (JVM).

JVM este un mediu de execuție necesar pentru rularea codului de octeți Spark.

Deoarece Spark este scris în Scala/Java:

• Când se folosește Scala sau Java în aplicație: Codul rulează direct în JVM.
• Când se folosește PySpark (Python): Un proces Python rulează codul, dar comunică cu un driver bazat pe JVM printr-o punte precum Py4J. Executorii Spark de pe nodurile worker rulează în continuare într-un JVM, care gestionează apoi workerii Python pentru a executa cod Python și funcții definite de utilizator în Python (Python UDF-uri).

În ambele cazuri, JVM gestionează firele de execuție, alocarea memoriei și colectarea gunoiului atât pentru driver, cât și pentru executori.

JVM-urile driver și executor dintr-o aplicație PySpark

Într-o aplicație PySpark, driverul Python comunică cu driverul Spark care rulează în JVM;

• Driverul Python: Aici este scris și executat codul PySpark la nivel de aplicație.
• JVM-ul driverului: Coordonează execuția și programarea sarcinilor Spark.
• JVM-ul executorului: Rulează sarcinile Spark și gestionează resursele de execuție. Procesele worker Python sunt implicate atunci când sarcinile necesită execuție în Python.
• JVM-ul clientului (procesul client sau mașina de pe care se trimite aplicația): JVM care rulează pe mașina locală sau pe un nod edge/gateway.

După cum se arată în figură, JVM-ul clientului se conectează mai întâi la JVM-ul driverului, care coordonează apoi activitatea JVM-urilor executorilor:

Figura D. Relația generală dintre JVM-urile clientului, driverului și executorilor

De aici, JVM-ul driverului găzduiește driverul Spark, care gestionează aplicația, în timp ce JVM-urile executorilor găzduiesc executorii individuali responsabili de executarea acelor sarcini.

Fiecare driver JVM dispune de memorie heap JVM pentru a gestiona eficient procesarea datelor, memorarea în cache și obiectele utilizatorului.

2-4. Analiză aprofundată a seturilor de date distribuite reziliente (RDD)

Un set de date distribuit rezilient (RDD) este o colecție imuabilă, partiționată, de elemente pe care Spark o poate procesa în paralel.

RDD-urile urmăresc dependențele dintre transformări, permițându-i lui Spark să recalculeze partițiile pierdute în multe scenarii de defectare.

Denumirea reflectă trei proprietăți:

• Rezilient: Spark înregistrează linia de proveniență — ce RDD depinde de fiecare RDD anterior — și poate adesea să reconstruiască partițiile lipsă, în loc să păstreze replici ale fiecărui rezultat intermediar.
• Distribuit: Setul de date este reprezentat prin partiții pe care Spark le poate procesa pe executori diferiți. Nu sunt necesare mai multe mașini fizice; și modul local acceptă RDD-uri.
• Set de date: Un RDD poate conține perechi cheie-valoare, numere întregi, șiruri de caractere, tupluri și alte elemente acceptate — nu doar rânduri relaționale. API-urile RDD includ operațiuni precum map, filter, reduceByKey, join, și aggregateByKey.

Un RDD este o abstractizare logică a unei colecții și a calculului, nu o declarație că anumite partiții sunt atribuite permanent anumitor executori.

Partiții

O partiție este o diviziune logică a datelor și corespunde, de regulă, unei singure sarcini dintr-o etapă.

Să presupunem că creăm 100 de partiții (aceste 100 de partiții inițiale reprezintă o alegere didactică explicită).

Cu 10.000 de elemente sursă, rezultă aproximativ 100 de elemente per partiție dacă datele de intrare sunt distribuite relativ uniform:

Figura E. Alocarea paralelă a partițiilor în 100 de diviziuni inițiale ale datelor

3. API-ul RDD vs. API-ul DataFrame: optimizarea internă Catalyst

Exemplul original cu RDD ajută la înțelegerea dependențelor, a partițiilor și a redistribuirii datelor.

În ETL-ul structurat obișnuit, însă, API-ul DataFrame este punctul de pornire preferat, deoarece oferă coloane, informații despre schemă, expresii de tip SQL și operații vizibile pentru optimizator.

Diagrama de mai jos ilustrează relația:

Figura F. Relația dintre conceptele de graf din Spark: dependențele RDD, DataFrame și DAG-ul de execuție Spark

Așa cum RDD-ul are dependențe care definesc transformările, planul de interogare al unui DataFrame reprezintă interogarea sa structurată.

Spark construiește fluxul de execuție pe baza dependențelor și a operatorilor fizici.

Execuția DataFrame poate folosi intern mecanisme bazate pe RDD, dar asta nu înseamnă că utilizatorii trebuie să construiască manual dependențele RDD atunci când scriu cod pentru DataFrame.

3-1. Planul de interogare DataFrame

Un plan de interogare DataFrame este succesiunea de etape logice și fizice generată de motorul de execuție (caseta mov din Figura L).

O interogare DataFrame trece prin mai multe reprezentări:

1. Plan logic: Descrierea operațiilor solicitate (scanare, filtrare, grupare, agregare).
2. Plan logic analizat și optimizat: Spark rezolvă referințele și aplică optimizările logice relevante, precum eliminarea coloanelor inutile sau împingerea predicatelor.
3. Plan fizic: Spark selectează operatorii executabili, de exemplu scanări, agregări parțiale/finale, schimburi de date, îmbinări și sortări.
4. Execuție: Planul fizic este rulat folosind mecanismele de execuție Spark și implică, în cele din urmă, sarcini pe executori.

În cazul nostru de utilizare, planul fizic poate include următoarele etape:

• Agregare hash parțială,
• Schimb de date după SKU și
• Agregare hash finală.

3-2. Compararea acțiunilor: API-ul RDD vs. API-ul DataFrame

Să comparăm API-ul RDD și API-ul DataFrame, folosind drept exemplu agregarea după SKU.

API-ul RDD

Elementele sunt (sku, qty) perechi. Logica noastră de agregare în Python este explicită, iar Spark urmează dependențele RDD:

result_rdd = (
    raw_rdd
    .filter(lambda row: row[1] > 0)
    .reduceByKey(lambda a, b: a + b, numPartitions=4)
)

rdd_results = result_rdd.collect()

# Example:
# [('A', 8000), ('B', 10500), ('C', 9200)]

API-ul DataFrame

Să presupunem că sales_df are coloanele sku (șir de caractere) și qty (numeric):

from pyspark.sql import functions as F

result_df = (
    sales_df
    .filter(F.col("qty") > 0)
    .groupBy("sku")
    .agg(F.sum("qty").alias("total_qty"))
)

df_results = result_df.collect()

# Example:
# [Row(sku='A', total_qty=8000),
#  Row(sku='B', total_qty=10500),
#  Row(sku='C', total_qty=9200)]

Aici, ambele exemple rezolvă aceeași problemă de business:

Filtrarea înregistrărilor de vânzări nevalide și agregarea cantităților valide după SKU.

Și, în ambele cazuri, collect() declanșează execuția Spark și returnează toate înregistrările rezultate către Driver.

Principala diferență ține de reprezentarea rezultatului:

• API-ul RDD: Returnează elemente Python, precum (sku, total_qty) tupluri.
• API-ul DataFrame: Returnează obiecte structurate Row cu câmpuri denumite, precum sku și total_qty.

Iată o comparație generală între API-urile RDD și DataFrame:

Tabelul 2. API-ul RDD vs. API-ul DataFrame – comparație între funcționalități și nivelul de abstractizare: (Creat de Kuriko IWAI)

4. Concluzii

Scopul optimizării Spark nu este maximizarea independentă a numărului de partiții sau a utilizării procesorului.

Scopul este reducerea costurilor și a latenței de execuție de la un capăt la altul, fără a provoca instabilitate.

În acest articol, aflăm care sunt relațiile fundamentale din sistemul Spark:

• Resurse: Lucrătorii găzduiesc executori; executorii rulează sarcini; numărul configurat de procesoare și memoria configurată limitează concurența.
• Programarea cu RDD: Transformările generează definiții imuabile ale RDD-urilor și informații despre proveniența acestora; acțiunile declanșează calculul.
• Programarea cu DataFrame: Transformările generează operații logice de interogare; optimizarea selectează operatorii fizici executabili; acțiunile declanșează calculul.
• Execuție: Spark organizează activitatea în joburi, etape și sarcini, iar sarcinile procesează partiții pe executori. Amestecurile de date și alte dependențe creează limite de coordonare între etape.

Înțelegerea modului în care Spark execută sarcini distribuite este doar punctul de plecare.

Poți afla despre un flux de calcul în șapte pași folosind API-ul RDD.