Narcis Miclaus

10 min di lettura

PySpark, dalle fondamenta Lezione 12 / 60

Local contro cluster: workflow di sviluppo che non ti mente

Quando local mode basta, quando ti serve un cluster vero, e i bug che si presentano solo quando ci sono executor veri in scena.

Questa è la lezione di chiusura del modulo 2. Tutto quello che hai scritto finora ha girato su master("local[*]"). È stato deliberato: local mode è davvero buono per imparare Spark, e la maggior parte di quello che hai digitato girerà senza modifiche su un cluster da 100 nodi. Ma “la maggior parte” sta facendo molto lavoro in quella frase. C’è una categoria di bug che esiste solo quando Spark è davvero distribuito, e scoprirli in produzione è un brutto momento.

Obiettivo di oggi: capire esattamente cosa significa local, su cosa ti mente, e come si presenta un workflow sano dev -> staging -> prod nel 2026.

Cos’è davvero local mode

Quando scrivi:

spark = SparkSession.builder.master("local[*]").getOrCreate()

…Spark avvia un singolo processo JVM sul tuo portatile. Quel processo è driver, executor e cluster manager tutto insieme. Niente rete. Nessuna serializzazione tra macchine. Nessun executor JVM separato. Solo un processo che finge di essere un cluster.

Il [*] in local[*] controlla il parallelismo: dice a Spark di usare tutti i core CPU disponibili come thread worker. Altre forme:

.master("local")       # 1 thread. Utile per test in cui vuoi determinismo.
.master("local[4]")    # 4 thread.
.master("local[*]")    # Tutti i core.
.master("local[*, 3]") # Tutti i core, ritenta i task falliti 3 volte. Testa la fault tolerance.

Ogni “thread” esegue un task Spark alla volta. Con local[8], puoi eseguire 8 partizioni in parallelo, su 8 core, in una JVM. Sembra un cluster dal punto di vista delle API. Il codice DataFrame che scrivi è byte-per-byte identico al codice cluster. Quella è la parte davvero utile.

Demo veloce: stesso script, eseguito due volte con parallelismo diverso:

from pyspark.sql import SparkSession
from pyspark.sql.functions import spark_partition_id, count

def run(master):
    spark = (SparkSession.builder
             .appName("LocalDemo")
             .master(master)
             .getOrCreate())
    spark.sparkContext.setLogLevel("WARN")

    df = spark.range(0, 1_000_000).repartition(16)
    print(f"\nMaster={master}, default parallelism={spark.sparkContext.defaultParallelism}")
    df.groupBy(spark_partition_id().alias("partition")).agg(count("*").alias("rows")).show()
    spark.stop()

run("local[1]")
run("local[*]")

Entrambi i run producono la stessa aggregazione finale. Il primo usa 1 thread worker, il secondo usa ogni core che hai. Il throughput differisce di ~Nx dove N è il tuo conteggio core. L’output è identico.

Dove local mode va davvero bene

Sii senza scuse riguardo a local per questi:

  • Iterazione di sviluppo. Scrivi una transformation, eseguila, aggiusta il type error, eseguila di nuovo. Il feedback loop su un cluster vero è 30 secondi di overhead di submission per tentativo. Local è istantaneo. Itererai dieci volte più velocemente.
  • Unit test. Una suite pytest che costruisce un DataFrame minuscolo nel codice, esegue la tua funzione e assert sull’output è esattamente ciò per cui local mode esiste. I CI runner hanno 2-4 core e ne basta.
  • ETL piccolo. Se il tuo input è sotto, diciamo, 5GB e il tuo portatile ha 16GB di RAM, local mode supera in performance un cluster a 4 nodi (niente rete, niente shuffle sul cavo, niente spin-up di executor). Ho una pipeline di finanza personale che gira ogni notte su local[*] e probabilmente lo farà sempre.
  • Apprendimento. Non imparerai le window function più velocemente su EMR che sul tuo portatile.

Un portatile con local[*] e 16GB di memoria è un ambiente Spark vero, utilizzabile. Non scusarti per usarlo.

Dove local mode ti mente

Tre categorie di bug che local mode nasconde. Tutte e tre prima o poi ti troveranno in produzione.

1. Lo skew è invisibile

In Spark vero, le partizioni sono distribuite tra gli executor. Se una partizione ha il 90% dei dati, mettiamo che il tuo groupBy("country") ha prodotto una partizione US con 50GB e altre quaranta partizioni con 100MB ciascuna, quella partizione US gira su un executor mentre gli altri 40 executor stanno fermi. Il wall-clock time del job diventa “quanto ci mette quel singolo task”. Questo è data skew, ed è la singola causa più comune di ticket “il mio job Spark è lento”.

Local mode non te lo mostra. Con un singolo processo, tutte le partizioni passano per la stessa JVM comunque, sugli stessi thread, condividendo la stessa memoria. Una partizione skewed ci mette comunque più tempo delle altre, ma può fare spill, swap e altrimenti aggirare il problema in-process. Su un cluster vero, quel singolo task ha esattamente la memoria di un executor e muore.

Concretamente:

from pyspark.sql.functions import lit, col
import random

# Costruisci un dataset pesantemente skewed: una key con 1M righe, altre con 1.
big   = spark.range(0, 1_000_000).withColumn("key", lit("HOT"))
small = spark.range(0, 99).withColumn("key", col("id").cast("string"))
skewed = big.unionByName(small)

# Group by key: un task fa quasi tutto il lavoro.
skewed.groupBy("key").count().show()

Su local[*] questo si completa felicemente. Su un cluster piccolo con impostazioni di default, il task della key HOT può mandare in OOM l’executor. La lezione 41 è dedicata allo skew (salting, broadcast join, gestione skew di AQE) e non puoi davvero esercitarti localmente.

2. Gli shuffle sembrano gratis

Uno shuffle in Spark è quando le partizioni vengono ridistribuite sulla rete: il motore dietro ogni groupBy, join, distinct e orderBy. Gli shuffle serializzano dati, li scrivono su disco locale, e li tirano via cavo dagli altri executor. Sono costosi. Sono spesso il collo di bottiglia. L’intera arte del “tunare Spark” è per lo più ridurre il volume di shuffle.

Localmente, uno shuffle è un trasferimento di memoria in-process. Niente rete. Spesso niente disco. È quasi gratis. Quindi localmente una pipeline a 5 shuffle gira in 8 secondi; su un cluster, la stessa pipeline ci mette 12 minuti. Gli shuffle dominano, e tu non l’avevi visto venire.

Inoltre non beccherai:

  • Timeout di rete ed executor persi durante le read di shuffle.
  • Failure per disco pieno da spill di shuffle.
  • Skew nell’output di shuffle (un reducer di shuffle con 50GB di dati).

Le pipeline che sembrano fantastiche su un portatile hanno una tendenza imbarazzante a cadere a pezzi su un cluster, e la causa è quasi sempre “gli shuffle a cui hai smesso di pensare perché erano gratis localmente”.

3. L’out-of-memory a scala di executor non si riproduce

In un cluster vero, ogni executor ha memoria limitata, mettiamo 16GB per executor. Se una partizione fa spill di 20GB in un singolo task, quel task muore. L’executor potrebbe morire. Il job fallisce.

Localmente, il tuo “executor” è la tua intera JVM, che può usare la maggior parte della RAM del portatile (spesso 8-32GB). Puoi processare partizioni che non starebbero in nessun executor di cluster ragionevole e mai accorgertene. Poi spedisci a un cluster e lo stesso codice va in OOM ogni due run.

La versione diagnostica: un parse from_json() su singolo record che alloca 200MB di struct annidate, eseguito su 100M record: localmente sono 20GB di churn di memoria totale, la JVM lo gestisce; su un cluster sono 200MB per qualunque cosa si infili nel modello di pressione di memoria di un executor e arrivederci.

Ci sono cugini più sottili. Broadcast join che “funzionano” perché tutto è in un processo in realtà devono essere broadcast su un cluster. UDF che funzionano localmente ma serializzano una closure da 50MB a ogni executor. Coercion di tipo PyArrow che fanno round-trip leggermente diversi quando attraversano confini di processo. Niente di tutto questo è mai un problema localmente.

Il workflow di sviluppo che li becca

La forma che sopravvive al contatto con la realtà:

1. Sviluppa su local[*]. Loop di iterazione stretto. Scrivi transformation, esegui unit test, fai lo scaffolding della pipeline.

2. Esegui su un piccolo cluster di staging, su una slice rappresentativa di dati. Due o tre executor, rete vera tra loro, una versione campionata o anonimizzata dei dati di produzione. È qui che lo skew si presenta, dove gli shuffle diventano costosi, dove la memoria-per-executor diventa un vincolo vero.

Un sample del 10% su un cluster vero è enormemente più utile di un run al 100% su local. Il punto non è il volume, è la distributedness. Se un cluster piccolo non si strozza, uno grande di solito non lo farà nemmeno.

3. Promuovi in produzione. Stesso codice, --master diverso e dati più grandi. A questo punto hai già debuggato il 95% dei bug solo-cluster.

Saltare il passo 2 è l’errore di workflow più comune sul campo. La gente passa direttamente da local a prod, becca un bug di skew in produzione, e passa una serata a fare rollback. Paga la tassa di staging. È poco.

Il panorama dei cluster manager, edizione 2026

Quando fai submit a un cluster vero, il cluster manager è la cosa che alloca executor per il tuo job. Spark ne supporta quattro:

YARN è il cluster manager Hadoop originale. Se la tua azienda fa girare un’installazione on-prem Cloudera/Hortonworks/legacy-Hadoop, sei su YARN. È maturo, battle-tested, e uniformemente odiato. Sta anche perdendo lentamente quote di mercato: praticamente nessuno sceglie YARN per un deployment greenfield nel 2026, ma la base installata è enorme e non se ne va in fretta.

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --num-executors 50 \
  --executor-cores 4 \
  --executor-memory 16g \
  --driver-memory 8g \
  my_pipeline.py

Kubernetes è il default moderno. Se stai tirando su Spark da zero nel 2026, è probabilmente su Kubernetes. Lo Spark Operator (originariamente di Google, ora un progetto Apache) ti dà CRD SparkApplication, scheduling nativo di pod, e integrazione pulita con il resto del tuo stack cloud-native. Quasi ogni vendor Spark cloud (Databricks, EMR Serverless, Dataproc Serverless, OpenShift) è ora Kubernetes sotto, anche se lo nascondono.

spark-submit \
  --master k8s://https://my-cluster.example.com:6443 \
  --deploy-mode cluster \
  --conf spark.kubernetes.container.image=my-registry/spark:3.5.1 \
  --conf spark.executor.instances=50 \
  --conf spark.kubernetes.namespace=data \
  local:///opt/jobs/my_pipeline.py

Standalone mode è il cluster manager bundled di Spark stesso. È un singolo binario, niente Hadoop o Kubernetes richiesti. Davvero utile quando hai un team piccolo e dedicato che fa girare Spark su un set fisso di VM e non vuoi imparare Kubernetes per quello. Il trade-off: niente isolamento di risorse tra i job (è first-come-first-served), niente condivisione di workload con job non-Spark, e una storia HA limitata. Per una box analytics da 5 macchine, va bene. Per una piattaforma multi-team, no.

# Sul nodo master
./sbin/start-master.sh
# Su ogni worker
./sbin/start-worker.sh spark://master-host:7077

# Submit
spark-submit \
  --master spark://master-host:7077 \
  --total-executor-cores 32 \
  --executor-memory 8g \
  my_pipeline.py

Mesos esisteva e dovresti sapere che esisteva. È stato deprecato in Spark 3.2 (2021) e rimosso interamente in Spark 4.0 (2024). Se qualcuno cerca di convincerti a fare deploy di Mesos nel 2026, gentilmente allontanalo.

I wrapper gestiti, ovvero Databricks, AWS EMR, Google Dataproc, Azure Synapse, sono tutti costruiti sopra uno di questi. Databricks gira su un substrato gestito di sapore Kubernetes. EMR può girare su YARN (EMR classico) o Kubernetes (EMR su EKS) o serverless. Dataproc ha la stessa divisione. Lo script di submit che scrivi cambia appena; ciò che cambia è chi gestisce i nodi sottostanti, chi li patcha, chi li paga.

Se sei agli inizi della tua carriera Spark, la risposta pratica per il 2026 è: impara local[*] per lo sviluppo, impara abbastanza Kubernetes da poter leggere un YAML SparkApplication, e capisci che servizi gestiti come Databricks sono ottimi per non dover pensare al resto. Prendi YARN se e quando un job lo richiede.

Un spark-submit minimale che puoi davvero usare

Ecco uno starter portabile. Lo stesso script funziona su local, standalone, YARN e Kubernetes: cambiano solo --master e poche righe --conf:

spark-submit \
  --master "local[*]" \
  --name "my_pipeline" \
  --conf "spark.sql.shuffle.partitions=200" \
  --conf "spark.sql.adaptive.enabled=true" \
  --py-files dependencies.zip \
  --files config.yaml \
  pipelines/my_pipeline.py \
  --input  s3a://my-bucket/raw/2026-05-01 \
  --output s3a://my-bucket/curated/2026-05-01

I pezzi:

  • --master: il cluster manager. local[*], yarn, k8s://..., o spark://....
  • --name: ciò che compare nella Spark UI. Dai un nome ai tuoi job. Il te del futuro che fa debug alle 3 di notte ringrazierà il te del presente.
  • --conf: override di configurazione Spark. I due qui sopra (shuffle.partitions e AQE) sono i più comuni.
  • --py-files: moduli Python extra zippati e spediti agli executor. Qualunque cosa oltre uno script in singolo file ne ha bisogno.
  • --files: file non-codice (config, dati di lookup) spediti nella working directory di ogni executor.
  • Il path dello script e qualsiasi argomento da inoltrare al tuo argparse.

Per un entry point Python, il blocco if __name__ == "__main__": del tuo script fa il parse degli argomenti, costruisce una SparkSession, esegue la pipeline, chiama spark.stop(), ed esce. Stesso file che lanceresti localmente con python my_pipeline.py, con gli stessi argomenti argparse. La portabilità viene dal fatto che lo script non si interessa di come è stato lanciato.

Un esercizio pratico che ripaga: prendi qualunque degli snippet PySpark multi-step di prima in questo modulo e trasformalo in un singolo pipelines/orders_etl.py submittabile. Leggi i path da argparse. spark = SparkSession.builder.appName("orders_etl").getOrCreate(): nessun master nel codice, quindi prende qualunque --master con cui lo submitti. Lancialo con spark-submit --master "local[*]" pipelines/orders_etl.py --input ./data/orders.csv --output ./data/orders.parquet. Una volta che funziona, lo stesso script girerà su un cluster con un solo flag cambiato.

Cosa ha coperto questo modulo

Modulo 2 fatto. Sai installare Spark, costruire una SparkSession, leggere CSV / JSON / Parquet, ispezionare con show / count / collect (con cura), scrivere con save mode e partitioning appropriati, e fare submit di un job a un cluster manager vero. Tutto da qui in poi è fare cose più interessanti con i DataFrame: select, filter, join, aggregate, window, e infine la roba di debug in produzione del modulo 10.

Prossimo modulo: lavorare davvero con i DataFrame. La Column API, select contro selectExpr, withColumn e le sue stranezze, e le piccole abitudini che separano il codice Spark che scala da quello che no.

Cerca