Narcis Miclaus

8 min de citit

PySpark, de la zero Lecția 55 / 60

Spark UI: cea mai importanta unealta pe care o vei invata

Un tur ghidat al fiecarui tab, Jobs, Stages, Tasks, SQL, Storage, Executors, si ce iti spune fiecare cand ceva nu merge.

Dacă înveți o singură unealtă operațională din tot cursul ăsta, fă să fie Spark UI. Orice întrebare de performanță, „de ce e lent jobul ăsta”, „de ce a făcut OOM”, „funcționează cache-ul meu”, „se face push-down la filtrul meu”, „de ce face un singur task toată treaba”, își are răspunsul în UI. Logurile îți spun ce s-a întâmplat. UI-ul îți spune de ce a fost lent, iar asta e întrebarea care plătește facturile.

Modulul 10 e jumătatea de producție a cursului. Îl vom petrece cu debugging, tuning și trusa de scule la care apelează un inginer Spark la 2 noaptea. Lecția asta e piesa centrală: UI-ul, fiecare tab, ce coloane să citești prima dată, cum arată tiparele.

Unde îl găsești

Pentru un SparkSession local, UI-ul se leagă implicit la localhost:4040. Dacă portul 4040 e ocupat (fiindcă mai rulează un alt driver), Spark incrementează: 4041, 4042 și așa mai departe. Driverul scrie în log URL-ul real la pornire. Poți întreba și sesiunea direct:

spark = SparkSession.builder.master("local[*]").getOrCreate()
print(spark.sparkContext.uiWebUrl)
# http://192.168.1.10:4040

Pe clustere reale, cluster manager-ul găzduiește (sau face link la) UI:

  • Databricks: fiecare pagină de cluster are un link „Spark UI”. Cât rulează clusterul, UI-ul e live; pentru clustere terminate, Databricks păstrează un randare statică din event log.
  • EMR: UI-ul YARN ResourceManager de pe nodul master listează fiecare aplicație; click pe aplicație, click pe „ApplicationMaster” pentru UI-ul live, sau „History” după ce se termină.
  • Kubernetes / Spark Operator: pod-ul de driver expune 4040 intern; faci kubectl port-forward la laptopul tău.
  • YARN simplu: la fel ca EMR, ResourceManager linkuiește spre el.
  • Spark History Server: pentru aplicații finalizate, un server separat citește event logurile (spark.eventLog.dir în care ai scris) și re-randează UI-ul la cerere.

UI-ul e o vedere read-only a stării interne a driverului Spark. Nu poți strica nimic apăsând pe butoane. Apasă pe tot.

Tab-urile în ordinea utilității

Spark expune un set fix de tab-uri. Le voi parcurge în ordinea în care le folosesc efectiv în producție, nu de la stânga la dreapta.

1. Jobs: vederea de ansamblu

Pagina de aterizare. Un rând per job, unde un job e tot ce e declanșat de o singură acțiune (.show(), .count(), .write(), .collect()). Coloane:

  • Job ID: secvențial.
  • Description: numele acțiunii sau ce ai setat cu spark.sparkContext.setJobDescription("...") (fă asta; e gratis și face UI-ul lizibil).
  • Submitted: timestamp wall-clock.
  • Duration: durata wall-clock.
  • Stages: Succeeded/Total: câte stages a avut jobul.
  • Tasks (for all stages): Succeeded/Total: numărul total de task-uri și câte sunt gata.

Folosește tab-ul Jobs ca să răspunzi la „ce acțiune e lentă”. Sortează după Duration. Jobul lent e cel în care intri.

Dacă vezi o descriere de job de tipul „count at NativeMethodAccessorImpl” fără alt context, ai un .count() undeva care declanșează ceva fără să-ți dai seama, deseori un debug print uitat în producție.

2. Stages: carnea

Faci click pe un job și obții stage-urile lui. Aici îți vei petrece majoritatea timpului.

Un stage e o secvență de operații pe care Spark le poate rula fără un shuffle. Fiecare shuffle e o graniță de stage. Un job tipic de join are trei stage-uri: scanează și pre-shuffle partea stângă, scanează și pre-shuffle partea dreaptă, fă joinul.

Coloane per stage de citit primele:

  • Duration: cât a durat stage-ul.
  • Tasks: Succeeded/Total: numărul de task-uri îți spune numărul de partiții.
  • Input: bytes citiți din sursă.
  • Output: bytes scriși.
  • Shuffle Read / Shuffle Write: bytes amestecați. Shuffle-urile mari sunt shuffle-uri scumpe.

Click pe un stage și obții Task table. E cel mai util ecran din tot UI-ul.

Pentru fiecare task vezi Status, Duration, GC Time, Shuffle Read Size, Shuffle Read Records, Spill (Memory), Spill (Disk) și Errors. Sortează după Duration descending. Acum uită-te la:

Max vs median. Dă click pe „Summary Metrics” sus, Spark îți dă min / 25% / median / 75% / max pentru fiecare coloană numerică. Dacă durata maximă a unui task e 10× mediana, ai skew. Dacă input-ul max e 10× mediana, ai partiții de input distorsionate. Dacă shuffle read max e 10× mediana, ai o cheie hot într-un join sau groupBy. Skew-ul e cea mai comună problemă în producție; tabela de task-uri e locul unde îl vezi.

Coloane spill. Spill (Memory) e cât de multă date necomprimate a trebuit Spark să scoată din memoria de execuție. Spill (Disk) e cât de multă date comprimate au ajuns pe discul local din cauza acelei împingeri. Orice spill înseamnă că ai rămas fără memorie de execuție și a trebuit să cazi pe disc. Puțin e normal sub presiune; mult e motivul pentru care jobul tău e lent. Vom vorbi despre cum se rezolvă în lecția 57.

GC time. Timp petrecut în garbage collection-ul JVM per task. Dacă GC e mai mult de ~10% din durata task-ului, ești sub presiune de memorie chiar dacă nu apare niciun spill. Crește memoria executor-ului sau redu datele per task.

Vizualizarea DAG din capul paginii de stage e și ea utilă, desenează operatorii din interiorul stage-ului și săgețile dintre ei. Utilă pentru a înțelege ce face stage-ul, mai puțin utilă pentru debugging de performanță decât tabela de task-uri.

3. SQL / DataFrame: tab-ul cu planul de query

E tab-ul cel mai click-uit în producție pentru workload-uri DataFrame și SQL. Fiecare query pe care îl rulezi apare ca un rând cu textul SQL original sau o descriere generată, plus un link „Execution ID”.

Click pe Execution ID și obții graful operatorilor: fiecare nod din planul fizic ca o cutie, fiecare cutie adnotată cu numere de rânduri, timp, output rows și bytes. Adnotările vin din rularea efectivă, nu din estimări, e vederea post-mortem a ce a făcut Spark.

La ce să te uiți:

  • Operatori Exchange neașteptați. Fiecare Exchange e un shuffle. Două Exchange-uri unde te așteptai la unul înseamnă că ai cauzat din greșeală un re-shuffle (deseori prin repartiționare dublă sau prin spargerea unei co-partiționări).
  • Cutia algoritmului de join. BroadcastHashJoin e rapid. SortMergeJoin e default-ul sigur. BroadcastNestedLoopJoin e un miros de cod; de obicei înseamnă că optimizatorul nu a putut alege o strategie din cauza unei condiții de join non-equi.
  • Numere de rânduri pe filtre. Dacă filtrul tău spune „1 miliard rânduri în, 1 miliard rânduri afară”, filtrul nu filtrează, de obicei din cauza unei nepotriviri de tip.
  • Adnotări Adaptive Query Execution. Cu AQE pornit, tab-ul SQL e singurul loc unde vezi ce s-a rulat de fapt, fiindcă planul s-a schimbat la runtime. Caută cutii AQEShuffleRead: e AQE care unește sau sparge partiții adaptiv.

Tab-ul SQL e și locul unde vei descoperi că ce a printat df.explain() înainte de rulare nu e ce s-a executat de fapt. Vom săpa în asta în lecția 56.

4. Storage: cel care spune adevărul despre cache

Tab-ul Storage listează fiecare DataFrame cached sau persisted, cu mărimea reală pe cluster, nivelul de stocare (memory only, memory and disk etc.), fracțiunea cached și distribuția per executor.

Ce să citești primul:

  • Fraction cached. Dacă nu e 100%, nu ai memorie suficientă și Spark a evictat partiții. Cache-ul e parțial rece, iar următoarea citire va fi recompute parțial.
  • Size in Memory. Cât e efectiv rezident, deseori mult mai mare decât mărimea Parquet pe disc din cauza formatului in-memory.
  • Nume RDD. Dacă codul tău apelează .cache() pe mai multe DataFrame-uri, dă-le nume umane mai întâi cu df.createOrReplaceTempView("nume") sau setează df.rdd.name = "..." ca să le poți distinge în UI.

Storage e cum confirmi că .cache() a făcut efectiv ce te așteptai. Dacă tab-ul e gol după acțiunea ta, apelul tău de cache nu a fost materializat, caching-ul e lazy, trebuie să declanșezi o acțiune care atinge DataFrame-ul cached ca să se populeze.

5. Executors: sănătatea per-executor

Un rând per executor (plus un rând pentru driver). Coloane:

  • Address: host:port.
  • Status: Active sau Dead.
  • RDD Blocks / Storage Memory: câtă date cached ține acest executor.
  • Disk Used: shuffle și spill data pe discul local.
  • Cores: cores asignate acestui executor.
  • Active / Failed / Complete Tasks: throughput de task-uri.
  • Task Time (GC Time): timp total de task și fracțiunea de GC.
  • Input / Shuffle Read / Shuffle Write: bytes procesați.

Cea mare: GC Time ca fracțiune din Task Time. Dacă un executor petrece 30% din timp în GC, se îneacă în obiecte și JVM-ul e gâtul de sticlă. Fie crește memoria, fie redu cores per executor, fie repară workload-ul. Presiunea GC e un ucigaș tăcut, jobul tău rulează, dar la jumătate din viteză și nimic nu loghează o eroare.

Celălalt lucru de verificat sunt Dead executors. Dacă executorii continuă să moară și să fie înlocuiți, e ceva în neregulă, de obicei OOM kills de la cluster manager. Click pe executor-ul mort, uită-te la log-ul executor-ului, găsește motivul kill-ului. Lecția 57 e lecția de OOM postmortem.

6. Streaming / Structured Streaming

Prezent doar dacă ai query-uri de streaming. Statistici per query: input rate, processing rate, batch duration, mărimea state operator-ului. Tiparele de urmărit sunt processing rate care rămâne în urma input rate-ului (nu poți ține pasul), batch duration care urcă în timp (state-ul crește nelimitat, problemă de watermark) și mărimea state operator-ului care atinge zeci de GB (ții prea mult state, e momentul să adaugi un watermark sau să regândești cheile).

Am acoperit jumătatea de streaming în lecțiile 49-54; tab-ul ăsta e locul unde vei diagnostica versiunea de producție.

7. Environment

Fiecare config Spark, proprietate JVM, intrare classpath și config Hadoop activă pentru acest driver. Util când debughezi „de ce se comportă asta diferit în producție”, jumătate din timp răspunsul e un config pe care echipa de platformă l-a setat și pe care nu-l știai. Caută spark.sql.adaptive, spark.serializer, spark.executor.memory ca să începi.

Arborele de decizie „unde mă uit prima dată”

Te-au sunat. Pipeline-ul e lent. Parcurge UI-ul așa:

  1. Tab-ul Jobs. Sortează după durată. Jobul lent e cel pe care faci click.
  2. Click pe jobul lent → Stages. Sortează stage-urile după durată. Stage-ul lent e cel pe care faci click.
  3. Click pe stage-ul lent → Task table. Click pe „Summary Metrics”.
    • Durata max ≫ mediana? Skew. Repară cu salting sau cu AQE skew handling.
    • Numere mari de spill? Presiune de memorie. Crește memoria executor-ului sau redu mărimea partițiilor.
    • GC time mare? Tot presiune de memorie, deseori mai multe cores per executor decât poate susține heap-ul.
    • Totul echilibrat și pur și simplu lent? Ești CPU-bound, ai nevoie de mai multe cores sau de un query mai inteligent.
  4. Verificare încrucișată în tab-ul SQL. Găsește query-ul, uită-te la graful de operatori, verifică strategia de join și numerele de rânduri la fiecare filtru. Vei observa deseori „filtrul care n-a făcut nimic” sau „broadcast-ul care nu s-a întâmplat”.
  5. Tab-ul Executors. Vreunul mort? Vreun executor face toată treaba? GC time arată sănătos?

Asta e 90% din debugging-ul Spark. UI-ul îți spune care din aceste patru lucruri e greșit. Să citești UI-ul fluent e diferența între un inginer Spark și cineva care rulează Spark.

Lecția următoare: .explain() și cum să citești planul înainte să-l rulezi, ca UI-ul să confirme ce te așteptai în loc să fie o surpriză.

Caută