04 · Capire come Spark esegue

Segui un’operazione DataFrame dall’espressione Python fino a job, stage e task. La lezione è autonoma e usa dati sintetici, PySpark classico 4.x e Java 17 o successivo. Su Kaggle avvia una nuova sessione CPU. Se serve, installa PySpark in una cella di preparazione con %pip install "pyspark>=4.0,<4.1", poi usa Run All.

from tempfile import TemporaryDirectory
from uuid import uuid4

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (
    SparkSession.builder.master("local[2]")
    .appName("04-understand-spark-execution")
    .getOrCreate()
)
spark.conf.set("spark.sql.shuffle.partitions", "2")
sc = spark.sparkContext
tracker = sc.statusTracker()

sales = spark.range(12, numPartitions=2).select(
    F.col("id").alias("sale_id"),
    (F.col("id") % 3).alias("shop_id"),
    (F.col("id") + 1).alias("units"),
)

1. DataFrame

Un DataFrame è una tabella con colonne dotate di nome e tipo. Questo descrive vendite sintetiche. Spark può distribuire le righe e calcolare operazioni su di esse.

2. Partizione

Una partizione è un’unità di dati che Spark può elaborare separatamente. La sequenza di partenza richiede due partizioni. Qui usiamo una volta df.rdd.getNumPartitions() per contarle. La vista RDD e i controlli delle partizioni tornano nelle lezioni successive.

3. Trasformazione

Una trasformazione descrive un nuovo DataFrame a partire da uno esistente. L’esempio filtra le righe e usa withColumn() per aggiungere un campo derivato.

4. Trasformazione stretta

Con questi filtri e proiezioni, ogni partizione di output dipende soltanto dalla corrispondente partizione di input. Spark può elaborarla senza redistribuire le righe.

5. Trasformazione larga

Un’aggregazione per chiave può richiedere valori da più partizioni di input. L’esempio raggruppa per negozio e le vendite di uno stesso negozio compaiono in entrambe le partizioni.

6. Lineage

Il lineage descrive le dipendenze risalendo alle operazioni e agli input precedenti. Qui i totali per negozio dipendono dalle vendite preparate, che a loro volta dipendono dalla sequenza iniziale.

7. Immutabilità del DataFrame

Ogni trasformazione restituisce un altro DataFrame. L’originale conserva colonne e significato.

print("Source partitions:", sales.rdd.getNumPartitions())
prepared = sales.filter(F.col("units") >= 3).withColumn("adjusted_units", F.col("units") * 2)
totals = prepared.groupBy("shop_id").agg(F.sum("adjusted_units").alias("total"))
assert "adjusted_units" not in sales.columns
assert "adjusted_units" in prepared.columns

8. Azione

Un’azione richiede un risultato o un effetto. Qui collect() restituisce a Python i totali dei tre negozi.

9. Job

Un job è un calcolo pianificato composto da stage. Un’azione può avviare più job. Per esempio, Spark può eseguire lavoro aggiuntivo per adattare una query SQL durante l’esecuzione.

10. Valutazione pigra

Di norma, costruire una trasformazione registra che cosa calcolare. L’azione avvia l’esecuzione. L’esempio etichetta separatamente la costruzione della query e la relativa azione.

11. Contare i job avviati

Usa un gruppo di job univoco per isolare il lavoro avviato dall’esempio. Il tracker di stato riporta gli ID dei job recenti di cui è a conoscenza e può aggiornarsi con un ritardo. Osserva i conteggi senza presumere che ogni azione avvii un numero fisso di job.

def observe_jobs(label, operation):
    group = f"nb04-{uuid4()}"
    sc.setJobGroup(group, label)
    try:
        result = operation()
        ids = sorted(tracker.getJobIdsForGroup(group))
        print(f"{label}: {len(ids)} tracked job(s), IDs {ids}")
        return result, ids
    finally:
        for key in ("spark.jobGroup.id", "spark.job.description", "spark.job.interruptOnCancel"):
            sc.setLocalProperty(key, None)

query, _ = observe_jobs("Build query", lambda: totals.filter(F.col("total") > 0))
rows, action_ids = observe_jobs("Collect shop totals", query.collect)
assert {row.shop_id: row.total for row in rows} == {0: 42, 1: 48, 2: 60}

12. Azione dentro un ciclo

Contare separatamente per ogni negozio invia richieste di lavoro distinte. Un conteggio per gruppo esprime lo stesso calcolo in una query. Confronta i job rilevati e verifica che le due versioni restituiscano gli stessi conteggi.

def count_each_shop():
    return {shop: prepared.filter(F.col("shop_id") == shop).count() for shop in range(3)}

loop_counts, _ = observe_jobs("Count each shop separately", count_each_shop)
group_counts, _ = observe_jobs("Collect grouped counts", prepared.groupBy("shop_id").count().collect)
assert loop_counts == {row.shop_id: row["count"] for row in group_counts}

13. Rimescolamento

Un rimescolamento (shuffle) redistribuisce i dati tra partizioni. Per aggregare per negozio, Spark può prima calcolare totali parziali e poi scambiarli affinché quelli di ciascun negozio arrivino alla stessa destinazione. Cerca Exchange nel piano fisico.

14. Stage

Uno stage raggruppa task che possono partire quando gli input necessari sono pronti. In genere uno shuffle separa lo stage che produce i dati da quello che li consuma.

totals.explain()

15. Driver

Il driver coordina l’applicazione, prepara l’esecuzione e pianifica il lavoro. Il kernel del notebook è la parte Python del driver in questo esempio.

16. Executor

Un executor esegue task e può tenere i dati dell’applicazione in memoria o su disco. In un cluster, gli executor sono processi assegnati all’applicazione.

17. Task

Un task è un’unità di lavoro su una partizione all’interno di uno stage. Se ci sono slot disponibili, più task possono essere eseguiti contemporaneamente.

18. Esecuzione locale

Con local[2], Spark usa due thread di lavoro sulla stessa macchina. Puoi comunque osservare partizioni, stage e shuffle. Questa configurazione funziona senza un cluster separato.

19. Gestore del cluster

Un gestore del cluster assegna risorse alle applicazioni Spark. Tra gli esempi ci sono Spark Standalone, YARN e Kubernetes.

20. Rimozione di Mesos in Spark 4.0

I materiali meno recenti possono citare Mesos come gestore del cluster. Spark 4.0 ha rimosso il supporto a Mesos.

21. Gerarchia job, stage e task

Un job fa riferimento a degli stage e ogni stage contiene task. Più job possono condividere uno stage. Il tracker di stato può mostrare gli ID degli stage associati all’azione precedente e il numero di task per ogni stage noto.

settings = sc.getConf()
print("Master:", settings.get("spark.master"))
print("SQL shuffle partitions:", spark.conf.get("spark.sql.shuffle.partitions"))

for job_id in action_ids:
    job = tracker.getJobInfo(job_id)
    if job is not None:
        for stage_id in job.stageIds:
            stage = tracker.getStageInfo(stage_id)
            if stage is not None:
                print(f"Job {job_id}, stage {stage_id}, tasks {stage.numTasks}")

22. DAG

Un grafo orientato aciclico rappresenta le dipendenze. La direzione indica quale lavoro dipende da quello precedente. Il grafo è privo di cicli di dipendenza.

23. DAG scheduler e task scheduler

Il DAG scheduler organizza l’esecuzione in stage e li invia quando le loro dipendenze sono pronte. Il task scheduler assegna i task dello stage inviato agli slot disponibili.

24. Stage saltato

Spark può saltare uno stage se gli output richiesti esistono già, per esempio se sono ancora utilizzabili i dati di shuffle prodotti in precedenza. L’interfaccia può indicare lo stage come saltato. L’esempio funziona anche se non ne compare nessuno.

25. Mostrare il grafo delle dipendenze

Se l’interfaccia Spark è raggiungibile, apri la scheda Jobs o Stages e scegli un job o uno stage per ispezionarne il DAG. Un notebook ospitato può limitare l’accesso alla porta dell’interfaccia. La cella seguente stampa l’indirizzo, se disponibile, e una rappresentazione testuale delle dipendenze dell’RDD sottostante al DataFrame.

print("Spark UI:", sc.uiWebUrl)
debug_view = totals.rdd.toDebugString()
print(debug_view.decode("utf-8") if isinstance(debug_view, bytes) else debug_view)

26. Tolleranza ai guasti

Spark può ritentare i task falliti e ricalcolare le partizioni perse quando input e dipendenze sono disponibili. Il recupero dipende da che cosa è fallito, da quante volte è successo e dalla possibilità di leggere ancora i dati necessari.

27. Salvare un checkpoint

Un checkpoint affidabile di un DataFrame materializza il risultato in una directory e tronca il piano del DataFrame restituito. L’esempio usa una directory temporanea perché il lavoro è locale. Un cluster richiede una posizione condivisa adatta ai checkpoint.

28. Checkpoint e cache

La cache agevola le letture ripetute e di norma conserva le dipendenze originali per un eventuale ricalcolo. Un checkpoint affidabile sostituisce le dipendenze a monte del DataFrame restituito con il risultato materializzato. Osserva il piano del checkpoint e confronta i risultati. unpersist() libera i dati in cache.

cached = totals.cache()
try:
    assert cached.count() == 3
    print("Cached:", cached.is_cached)
finally:
    cached.unpersist()

with TemporaryDirectory(prefix="spark-nb04-") as checkpoint_dir:
    sc.setCheckpointDir(checkpoint_dir)
    checkpointed = totals.checkpoint(eager=True)
    checkpointed.explain()
    assert {row.shop_id: row.total for row in checkpointed.collect()} == {
        row.shop_id: row.total for row in rows
    }

29. Esecuzione speculativa

Quando è abilitata, Spark può avviare un tentativo aggiuntivo per un task lento che soddisfa i criteri. Questo può ridurre l’impatto dei task più lenti e richiede risorse aggiuntive. Spark usa il risultato di un tentativo riuscito. La lezione legge l’impostazione dell’applicazione da SparkConf e si limita a osservarla.

print("Speculation enabled:", sc.getConf().get("spark.speculation", "false"))

Verifica la comprensione

1. Perché il filtro può essere eseguito all’interno di ogni partizione di origine?
2. Perché raggruppare per negozio può richiedere uno scambio?
3. Come cambia il checkpoint le dipendenze del DataFrame restituito?

Risposte

1. Il filtro può decidere se mantenere una riga usando i valori già presenti in quella riga.
2. Le vendite di un negozio possono trovarsi in più partizioni. I totali parziali devono confluire prima che Spark calcoli quello finale.
3. Il DataFrame con checkpoint legge il risultato materializzato e può quindi fare a meno di rieseguire il piano a monte.