Narcis Miclaus

13 min di lettura

PySpark, dalle fondamenta Lezione 60 / 60

Un health check da 30 minuti su un cluster Spark che non hai mai visto

La checklist di chiusura: ti consegnano il portatile, hai tempo fino alle 17 per capire cosa non va.

Ti consegnano le chiavi di un cluster Spark che non hai mai visto. “Dicci cosa non va entro le 17.”

Questa lezione è la checklist da 30 minuti che faccio io. Ogni passo, in ordine, con cosa stai cercando e cosa fare se lo trovi. È la lezione di chiusura del corso perché mette insieme tutto quello che abbiamo coperto (partitioning, shuffle, join, memoria, AQE, streaming, il tab SQL, il tab executors) e lo piega in un’unica sweep ordinata. La puoi stampare. La puoi incollare nella tua cassetta degli attrezzi. Il giorno in cui un nuovo cliente ti consegna un cluster Spark, la tiri fuori e in mezz’ora sei produttivo.

Grazie per essere arrivato fino alla lezione 60. Chiudiamola con una corsa scriptata attraverso l’intero toolkit.

Step 1: la panoramica del cluster (2 minuti)

Apri la Spark UI (o la UI del workspace se sei su Databricks/EMR/Dataproc). Trova l’applicazione in esecuzione. In alto a destra c’è la versione. Il tab Environment ti dà tutto il resto.

spark.version                              # 3.5.1, 3.4.2, etc.
spark.sparkContext.master                  # yarn, k8s://..., local[*], spark://...
spark.sparkContext.defaultParallelism      # total cores
sc = spark.sparkContext
print(sc.statusTracker().getExecutorInfos())

Stai cercando:

  • Versione di Spark: idealmente 3.2+, per AQE attivo di default. 3.0 / 3.1 vuol dire audit manuale delle config.
  • Cluster manager: YARN, Kubernetes, Standalone o vendor-managed. Strumenti operativi diversi.
  • Numero e dimensione degli executor: quanti, quanta memoria ciascuno, quanti core ciascuno. Confronta con la capacità di cluster annunciata.
  • Dimensione del driver: driver sottodimensionati crashano sui collect() grandi.

Step 2: il tab Executors (3 minuti)

Questo è il tuo sp_Blitz per Spark. Clicca Executors.

Executor ID | Address       | Status   | Cores | Memory Used | Task Time | GC Time | Failed | Active Tasks
driver      | 10.0.1.4      | Active   | 0     | 0 B / 4 GB  | 0 ms      | 0 ms    | 0      | 0
0           | 10.0.2.10     | Active   | 4     | 8.2 GB / 12 GB | 2.1 h | 18 min  | 0      | 4
1           | 10.0.2.11     | DEAD     | 4     | -           | 1.4 h     | 22 min  | 17     | 0
2           | 10.0.2.12     | Active   | 4     | 11.8 GB / 12 GB | 2.4 h | 1.1 h | 12     | 4

Stai cercando:

  • Executor morti. Se ne vedi, guarda i log (link stderr): OOM, nodo perso o il driver li ha killati per idle timeout. Morti ricorrenti = il tuo job è instabile.
  • GC time come frazione del task time. L’executor 2 sopra ha GC al circa 45% del task time. È catastrofico; la lezione 57 (memory tuning) è la fix.
  • Numero di task falliti. Diverso da zero su molti executor = problema reale (probabilmente skew o OOM); concentrato su un executor = nodo malato.
  • Memory usage vicino al limite. Executor a 11.8 / 12 GB sono a un’allocazione dallo spill o dall’OOM.

Annota chi ha brutta cera e vai avanti.

Step 3: job in esecuzione e query incastrate (2 minuti)

Clicca Jobs. Ordina per Duration. Tutto ciò che gira da più di 1 ora merita una domanda.

# In a notebook:
[(j.jobId, j.name, j.status, j.numTasks, j.numActiveTasks)
 for j in spark.sparkContext.statusTracker().getActiveJobIds()]

Stai cercando:

  • Job lunghi che dovrebbero essere veloci (un’aggregazione giornaliera che è girata 6 ore invece di 30 minuti, candidato al kill).
  • Job incastrati con un solo task attivo mentre tutti gli altri sono finiti: la classica coda di skew (lezione 28).
  • Stati incastrati lato driver: tanti stage completi ma il job non finisce; di solito un collect() o toPandas() che sta tirando giù troppa roba.

Per killare un job: la Spark UI ha un link (kill) accanto ai job in esecuzione se sei admin. Oppure spark.sparkContext.cancelJobGroup(...) se il tuo codice ha impostato un job group.

Step 4: statistiche cumulative dei job e i 10 più lenti (3 minuti)

Clicca di nuovo Jobs, scrolla giù alla lista Completed Jobs. Ordina per Duration decrescente. Guarda i primi 10.

Cosa hanno in comune?

  • Scrivono tutti sulla stessa tabella? -> il layout di storage di quella tabella potrebbe essere il collo di bottiglia (file piccoli, compaction mancante, partitioning sbagliato).
  • Fanno tutti join sulla stessa dimension? -> quella dim potrebbe aver bisogno di un broadcast (lezione 27) o di una chiave migliore.
  • Leggono tutti dalla stessa sorgente? -> controlla statistiche e partitioning di quella sorgente.
  • Girano tutti alla stessa ora del giorno? -> contesa di risorse con un altro job.

Riconoscere i pattern è più veloce che tunare ciascuno.

Step 5: uso del disco sui worker (3 minuti)

Le local dir di Spark (spark.local.dir, default /tmp) ospitano file di shuffle, file di spill, file di broadcast e block in cache. Si riempiono.

# On each worker (via SSH or kubectl exec):
df -h /tmp
du -sh /tmp/spark-* 2>/dev/null | sort -h | tail

Su piattaforme managed, controlla la metrica disco del worker nella UI della piattaforma. Su Databricks, è il pannello storage del cluster. Su EMR, le metriche disco di Ganglia o CloudWatch.

Stai cercando:

  • /tmp oltre l’80% pieno -> shuffle e spill falliranno. Aggiungi disco o riavvia il cluster.
  • Vecchie directory spark-*-shuffle da job falliti che non sono state pulite. Si puliscono allo shutdown graceful ma rimangono dopo i crash. È sicuro cancellarle dopo aver confermato che nessuna applicazione è in esecuzione.
  • File di cache che non sono stati puliti (directory blockmgr-*): stessa storia.

Step 6: le query più costose nel tab SQL (3 minuti)

Clicca SQL / DataFrame. Ordina per Duration. Le top 10 query sono dove sta andando il tempo del tuo cluster.

Per ognuna delle peggiori:

  • Cliccaci dentro. Guarda il piano.
  • Trova l’operatore peggiore (tempo riportato più alto).
  • Applica la checklist della lezione 58: skew? Strategia di join sbagliata? Shuffle massiccio? Filtro che non fa pushdown?

Colpevoli comuni su un cluster fresco:

  • Un BroadcastNestedLoopJoin da qualche parte: condizione di join mancante.
  • Un SortMergeJoin tra una tabella da 50 GB e una lookup da 5 MB perché qualcuno ha disabilitato autoBroadcastJoinThreshold anni fa e se l’è dimenticato.
  • Un groupBy con 50 partition da 8 GB ciascuna perché AQE è spento.
  • Python UDF (BatchEvalPython) che fanno un lavoro che potrebbe essere una funzione SQL (lezione 40).

Queste sono le query da archiviare come follow-up, non necessariamente da sistemare oggi.

Step 7: fallimenti recenti e classe di errore (2 minuti)

Per cluster Databricks: il tab Event Log e Failed Jobs. Per YARN: yarn application -list -appStates FAILED e i log dell’applicazione. Per Kubernetes: kubectl get pods e i pod degli executor che sono usciti con codice diverso da zero.

Raggruppa i fallimenti recenti per classe di errore:

  • Driver OOM: java.lang.OutOfMemoryError: Java heap space sul driver. Di solito un collect(), toPandas() o un broadcast enorme. Lezione 57.
  • Executor OOM: stessa eccezione su un executor. Skew (lezione 28), caching sbagliato (lezione 23) o memoria sottodimensionata.
  • Lost executor: ExecutorLostFailure seguito da retry dello stage. Di solito killato per OOM dall’OS o dal container manager. Alza memoryOverhead.
  • Shuffle fetch failed: FetchFailedException. Un executor è morto mentre un altro stava leggendo da lui. Sintomo di una delle due cose sopra.
  • Task failed N times: hai colpito il tuo spark.task.maxFailures (default 4). Bug reale o skew persistente.

La ripartizione ti dice la classe dominante, che ti dice quale fix prioritizzare.

Step 8: query streaming e checkpoint (3 minuti)

Se qualcuno fa girare Structured Streaming su questo cluster:

for q in spark.streams.active:
    last = q.lastProgress
    print(q.name, q.status, last.get("inputRowsPerSecond"), last.get("processedRowsPerSecond"),
          last.get("batchDuration"), last.get("triggerExecution"))

Stai cercando:

  • Query in cui input rate > processed rate in modo sostenuto -> stanno restando indietro, backlog illimitato.
  • Query con batchDuration > intervallo di trigger -> non riescono a stare dietro; cluster più grande o workload più piccolo.
  • Query in cui numInputRows è zero da ore -> upstream morto, query inattiva.

Poi controlla lo storage dei checkpoint:

# S3 or DBFS or whatever your checkpoint root is
aws s3 ls s3://my-checkpoints/ --recursive --summarize | tail -3

Stai cercando:

  • Directory di checkpoint di query morte che nessuno ha pulito: spreco di storage.
  • Singoli alberi di checkpoint dell’ordine dei GB perché la retention policy non sta facendo pruning dello stato. Le query stateful (lezione 53) senza watermark accumulano per sempre.

Step 9: job schedulati e fallimenti recenti (2 minuti)

I cluster manager hanno tutti viste sui job schedulati.

  • Databricks: tab Workflows -> Jobs -> ordina per stato dell’ultima run.
  • EMR: stato degli step; run dei DAG Airflow se orchestrate così.
  • Dataproc: workflow template e run recenti.
  • Plain Airflow / Dagster / etc.: task instance fallite nell’orchestratore.

Stai cercando:

  • Run fallite negli ultimi 7 giorni. Pattern: stesso job che fallisce ogni giorno, stessa ora del giorno, stesso errore?
  • Job che non girano da mesi. Schedule disabilitato? Vanno cancellati?
  • Run riuscite ma 4x più lente del solito -> regressioni silenziose, archivia come follow-up.

Step 10: audit della cache (2 minuti)

Clicca il tab Storage.

RDD Name                | Storage Level    | Cached Partitions | Memory   | Disk
df_users (id 12)        | MEMORY_AND_DISK  | 200 / 200         | 18 GB    | 0
df_huge_facts (id 14)   | MEMORY_ONLY      | 80 / 1200         | 12 GB    | 0
df_unused_2024 (id 5)   | MEMORY_AND_DISK  | 200 / 200         | 6 GB     | 0

Stai cercando:

  • Dataset in cache più grandi di quanto ci stia in memoria (la seconda riga sopra: solo 80 di 1200 partition in cache). La cache non sta facendo nulla; o alzi la memoria o smetti di cachare.
  • Dataset in cache che non sono stati toccati da ore: leakati da un notebook ancora attaccato. Fai unpersist.
  • Più DataFrame in cache molto simili: qualcuno ha cachato gli stessi dati tre volte sotto nomi di variabili diversi.

spark.sparkContext._jsc.getPersistentRDDs() elenca tutto quello che è attualmente in cache se vuoi scriptare l’audit.

Step 11: manutenzione tabelle Delta / Iceberg (2 minuti)

Se il cluster legge/scrive tabelle Delta o Iceberg, controlla che la manutenzione stia girando:

DESCRIBE HISTORY my.table LIMIT 10;     -- last operations
DESCRIBE DETAIL my.table;                -- numFiles, sizeInBytes

Stai cercando:

  • numFiles > 10.000 su tabelle sotto i pochi TB -> problema di file piccoli, serve OPTIMIZE.
  • Niente OPTIMIZE nella history da mesi -> schedula compaction settimanale.
  • Niente VACUUM neppure -> vecchi file versionati che si accumulano nell’object storage; costa soldi.
  • Equivalenti Iceberg: expire_snapshots e rewrite_data_files, anche questi schedulati.

Senza questo, le tue tabelle degradano in silenzio.

Step 12: sanity check sulla configurazione di memoria (2 minuti)

sc.getConf().getAll()

Oppure semplicemente il tab Environment nella UI, scrolla fino a Spark Properties.

Stai cercando:

  • spark.executor.memory e spark.executor.memoryOverhead: l’overhead dovrebbe essere all’incirca il 10% della executor memory o 1 GB, il maggiore dei due. Più stretto di così e YARN/K8s killeranno i container (lezione 57).
  • spark.driver.memory: se qualcuno chiama collect() su questo cluster, conta. Il default da 1 GB è troppo piccolo per qualunque cosa seria.
  • spark.executor.cores: tipicamente 4-5. Più alto significa più contesa per la JVM, più basso significa più JVM (overhead).
  • spark.sql.shuffle.partitions: se è 200 (il default) e stai processando terabyte, hai un problema. AQE aiuta ma non sistema tutto.

Se i container vengono killati (controlla la lista executor allo step 2 più gli eventi YARN/K8s), memoryOverhead è la prima cosa da alzare.

Step 13: configuration drift (2 minuti)

Stesso tab Environment. Filtra per spark.sql.* e spark.shuffle.*. Scansiona i valori non default.

Stai cercando:

  • spark.sql.adaptive.enabled = false -> riaccendilo (lezione 59) a meno che qualcuno non abbia un motivo documentato.
  • spark.sql.autoBroadcastJoinThreshold = -1 -> broadcast disabilitati. Quasi mai una buona idea su un workload reale.
  • spark.sql.shuffle.partitions = 2000 -> impostato alto una volta per un job spot, mai resettato. Ora colpisce ogni job sul cluster.
  • Serializer custom, codec, evict policy: ogni non-default ha bisogno di un commento nei docs del team che spieghi perché. Se nessuno sa perché, sospetto.

Documenta ogni non-default con la sua giustificazione. Tutto quello che non riesci a giustificare, ribalta al default.

Step 14: scrivere il report (3 minuti)

Prendi le tue note e trasformale in un report breve. Specchia il template in stile SQL Server: tienilo corto, prioritizzato, azionabile.

Health Check del cluster Spark del Cliente X, 2026-06-18

  • Overall: giallo. Due finding P1.
  • Critical:
    • La query streaming clickstream-aggs è 12 ore indietro; input rate 4x del processed rate. O cluster più grosso o ridurre il workload. Decidere oggi.
    • Cluster su Spark 3.0 -> AQE spento di default, più job ne beneficerebbero subito. Pianificare un upgrade o backportare le config AQE.
  • High:
    • OOM degli executor ogni giorno sul job daily-rollup. Segnale: skew su country_code. Salt o AQE.
    • 4 di 12 executor con GC time > 30% del task time. Alzare la memoria executor o ridurre i dataset in cache.
    • Mai eseguito OPTIMIZE su events_delta; 47.000 file piccoli su una tabella da 800 GB. Schedulare OPTIMIZE settimanale.
  • Medium:
    • BroadcastNestedLoopJoin nella query più costosa (weekly_finance). Condizione di join mancante; la query dura 40 minuti per run.
    • spark.sql.shuffle.partitions = 4000 impostato per un job spot a marzo, mai ripristinato; danneggia i job piccoli.
    • df_unused_2024 in cache (6 GB, MEMORY_AND_DISK) attaccato a un notebook inattivo. Fai unpersist.
  • Low:
    • 12 log di executor morti che si accumulano in /tmp sui worker. Aggiungere un cron di cleanup.
    • Il job etl-archive non gira con successo da gennaio. Disabilitare o sistemare.
  • Follow-up:
    • Audit di tutte le config Spark contro i default; documentare quelle da tenere.
    • Migrare il cluster a Spark 3.5 LTS.
    • Impostare uno schedule di OPTIMIZE + VACUUM Delta su tutte le tabelle di produzione.
    • Aggiungere dashboard Prometheus / DataDog a livello di cluster per GC degli executor, volume di shuffle, tasso di fallimento dei task.

Quello è il tuo deliverable. Corto, azionabile, prioritizzato. Mandalo via email a chiunque ti abbia consegnato le chiavi del cluster.

Step 15: cosa fare dopo questo corso

Ora sai l’equivalente di un paio d’anni di esperienza Spark sul campo, spremuti in 60 lezioni. Cosa leggere e seguire dopo:

  1. La documentazione di Spark: Performance Tuning e la sezione AQE. La fonte di verità per ogni config che toccherai.
  2. Il gitbook di Jacek Laskowski: The Internals of Apache Spark e The Internals of Spark SQL. Gratis, profondamente tecnico, il posto dove andare quando hai bisogno di sapere cosa fa davvero un operatore.
  3. I libri e i talk di Holden Karau: High Performance Spark e Learning Spark. Orientati alla produzione, scritti da qualcuno che fa debug di cluster Spark veri da anni.
  4. Il blog di Databricks: di parte verso la loro piattaforma, ma pieno di deep dive su Catalyst, AQE, Delta, Photon. Leggilo col filtro del bias acceso.
  5. La JIRA di Apache Spark: quando colpisci un bug strano, cercala. C’è un 50/50 che la fix sia nella prossima minor version.
  6. Pratica su un workload reale: una sandbox con TPC-DS o i tuoi dati. Lancia l’health check, sistema i finding, ripeti ogni mese. È così che si forma la memoria muscolare.

Congratulazioni

Ce l’hai fatta attraverso 60 lezioni di PySpark. Ora sai:

  • I fondamentali: cos’è Spark, l’architettura, la gerarchia RDD/DataFrame/Dataset (Moduli 1-2).
  • L’API DataFrame: read, write, transform, aggregate, window, pivot, UDF (Moduli 3-4).
  • La meccanica di esecuzione: lazy evaluation, narrow vs wide, il DAG, caching, shuffle, join, broadcast, skew, salting (Moduli 5-6).
  • Partitioning, bucketing, layout dei file, on-disk vs in-memory (Modulo 7).
  • L’optimizer: Catalyst, Tungsten, il tab SQL, formati di file, JDBC e cloud storage (Modulo 8).
  • Streaming: source, watermark, operazioni stateful, output mode, sink (Modulo 9).
  • Produzione: debug di job lenti, AQE, l’health check del cluster (Modulo 10).

Se questo corso ha fatto il suo lavoro, sei pronto ad aprire un cluster Spark che non hai mai visto, diagnosticare i suoi problemi in 30 minuti, sistemare i top issue e non andare nel panico quando il pager dell’on-call suona alle 3 di notte. Quella è l’asticella. Quello è cosa significa “sapere PySpark” a questo livello.

Continuerai a imparare cose nuove ogni settimana: è il mestiere. Ma ora hai il framework. Il prossimo blog post o item nelle release note si infila in una struttura che ha senso, invece di essere un muro di termini sconosciuti.

Vai a curarti delle tue pipeline. Si stanno appoggiando a te.

Grazie per aver letto. Se hai trovato errori, hai suggerimenti o vuoi discutere se sia giusto chiamare repartition() o coalesce(), fatti vivo.

Cerca