Cosa succede davvero quando si esegue un job Spark?

Introduzione

Con la crescita dei set di dati, elaborarli in modo efficiente su una singola macchina può diventare difficile a causa dei limiti della capacità della CPU, della memoria e della velocità di trasferimento dei dati dello storage.

Apache Spark affronta queste sfide offrendo un motore di elaborazione distribuita dei dati, in grado di suddividere i carichi di lavoro in partizioni ed eseguire le attività in parallelo su più core della CPU e su più macchine.

Tuttavia, Spark non consiste semplicemente nell’aggiungere altre macchine.

Per progettare pipeline di dati efficienti e affidabili è fondamentale capire come interagiscono i suoi componenti: driver, executor, RDD, DataFrame, partizioni, job, stage e task.

In questo articolo illustrerò l’architettura di Spark, i meccanismi di esecuzione e i principali fattori che incidono sulle prestazioni, usando un esempio pratico di aggregazione delle vendite di un e-commerce.

1. Perché Apache Spark: elaborazione parallela dei dati

Apache Spark è un motore di calcolo distribuito progettato per l’elaborazione di dati su larga scala e per l’ingegneria dei dati.

Spark elabora i big data sfruttando l’elaborazione parallela dei dati:

Figura A. Architettura di alto livello di un cluster distribuito Apache Spark (cluster, driver e nodi worker), con un confronto con l’esecuzione su un singolo nodo (creata da Kuriko IWAI)

Rispetto all’elaborazione su un singolo nodo (a destra, Figura A), Spark suddivide il lavoro in task utilizzando più nodi ed esegue le attività sulle risorse disponibili, distribuendole tra le macchine.

In questo caso, il termine distribuito non significa che siano obbligatorie più macchine. Spark può essere eseguito in modalità locale su una singola macchina. Anche in questo caso, i dati possono essere suddivisi in partizioni ed elaborati in parallelo.

2. Architettura di base di Apache Spark

Spark è composto da diversi componenti fondamentali:

• Driver Spark (driver): il processo centrale che coordina l’applicazione Spark.
• Gestore del cluster: un servizio esterno responsabile dell’allocazione delle risorse hardware, come i core della CPU e la memoria.
• Nodi worker (worker): una macchina fisica o virtuale oppure un nodo di calcolo del cluster.
• Executor: processi worker distribuiti avviati sui nodi worker.
• Java Virtual Machine (JVM): l’ambiente di runtime necessario per eseguire il bytecode di Spark (poiché Spark è scritto in Scala/Java).

Per prima cosa, il driver coordina un’applicazione Spark e la trasforma in un grafo diretto aciclico (DAG) di task di esecuzione fisici, quindi li pianifica sugli executor.

Il gestore del cluster alloca quindi le risorse nell’ambito del sistema di gestione delle risorse dell’ambiente di distribuzione.

I nodi worker forniscono risorse di calcolo agli executor, che eseguono i task e conservano i dati durante la loro esecuzione.

Alcuni ambienti gestiti, come Databricks, utilizzano un executor per worker, ma questa non è una regola universale di Spark. A seconda della configurazione del cluster, un worker può ospitare più executor.

2-1. Caso d’uso e-commerce: struttura dei dati di input

Immaginiamo una piccola azienda di e-commerce.

Dispone di 10,000 record di vendita, ciascuno con un codice SKU (stock keeping unit) e la quantità venduta. Supponiamo che vi siano tre SKU possibili: A, B, e C.

Tabella 1. Esempio di suddivisione dei record di vendita e-commerce e di aggregazioni previste.

Il nostro obiettivo è calcolare le unità totali valide vendute per SKU:

1. Escludere i record il cui qty <= 0, quindi
2. Aggregare i record per SKU.

Nel set di dati completo a titolo esemplificativo, supponiamo che dopo il filtraggio restino 9,800 dei 10,000 record. I totali finali attesi da Spark sono:

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

2-2. Cluster Spark

Definirò un cluster semplificato per il nostro caso d’uso.

Nodi worker (worker)

Imposterò 3 nodi worker, ciascuno con un’istanza dotata di 8 vCPU e 32 GiB di RAM:

Figura B. Allocazione delle risorse tra i nodi worker (creata da Kuriko IWAI)

Nel sistema sono disponibili in totale 24 vCPU.

Executor Spark e allocazione di CPU/memoria

Supponiamo che io riservi:

• 7 executor.
• 2 core vCPU per ogni executor.
• 3GiB di memoria heap JVM per ogni executor.

L’applicazione Spark chiede al gestore del cluster di allocare risorse a questi 7 executor.

Il gestore del cluster decide quindi come allocare CPU e memoria in base all’utilizzo delle risorse di ciascun worker.

Per esempio, decide di creare 14 slot per le attività: 2 core vCPU per executor × 7 executor:

Figura C-1. Core vCPU, memoria e allocazione degli executor tra i worker, con uno sbilanciamento delle risorse nel Worker A

Si noti che i 3 GiB di memoria allocati all’executor non corrispondono alla sua memoria totale, perché è necessario riservare memoria aggiuntiva per i worker e per altre esecuzioni.

Supponendo che una singola attività richieda 1 vCPU (spark.task.cpus = 1), ogni worker alloca le proprie vCPU:

Figura C-2. Allocazione parallela delle vCPU tra i worker

Questo mette in evidenza due aspetti fondamentali:

1. Ogni worker ha un tasso di utilizzo della CPU specifico (nella Figura C-2, l’utilizzo del Worker A è del 75%, mentre quello dei Worker B e C resta al 50%).
2. Anche se il sistema dispone di 24 core vCPU (8 vCPU per worker × 3 worker), può utilizzarne solo 14; le altre 10 restano inutilizzate per queste attività.

2-3. La Java Virtual Machine (JVM) alla base di Spark

Apache Spark si basa sulla Java Virtual Machine (JVM).

La JVM è un ambiente di esecuzione necessario per eseguire il bytecode di Spark.

Poiché Spark è scritto in Scala/Java:

• Quando si usa Scala o Java nell’applicazione: il codice viene eseguito in modo nativo all’interno della JVM.
• Quando si usa PySpark (Python): un processo Python esegue il codice, ma comunica con un driver basato su JVM tramite un bridge come Py4J. Gli executor Spark sui nodi worker continuano a essere eseguiti all’interno di una JVM, che gestisce quindi i worker Python per eseguire il codice Python e le funzioni definite dall’utente in Python (Python UDF).

In entrambi i casi, la JVM gestisce i thread di esecuzione, l’allocazione della memoria e la garbage collection sia per il driver sia per gli executor.

JVM del driver e degli executor in un’applicazione PySpark

In un’applicazione PySpark, il driver Python comunica con il driver Spark in esecuzione nella JVM;

• Driver Python: è l’ambiente in cui viene scritto ed eseguito il codice PySpark a livello di applicazione.
• JVM del driver: coordina l’esecuzione e la pianificazione delle attività Spark.
• JVM dell’executor: esegue le attività Spark e gestisce le risorse di esecuzione. Quando le attività richiedono l’esecuzione di codice Python, intervengono processi worker Python.
• JVM client (processo client o macchina da cui viene inviata l’applicazione): JVM in esecuzione sul computer locale o su un nodo edge/gateway.

Come mostra la figura, la JVM client si connette per prima alla JVM del driver, che coordina poi le JVM degli executor:

Figura D. Relazione complessiva tra le JVM client, del driver e degli executor

A questo punto, la JVM del driver ospita il driver Spark per gestire l’applicazione, mentre le JVM degli executor ospitano i singoli executor che eseguono le attività.

Ogni JVM del driver dispone di memoria heap JVM per gestire in modo efficiente l’elaborazione dei dati, la memorizzazione nella cache e gli oggetti utente.

2-4. Approfondimento sui Resilient Distributed Dataset (RDD)

Un Resilient Distributed Dataset (RDD) è una raccolta immutabile e partizionata di elementi che Spark può elaborare in parallelo.

Gli RDD tengono traccia delle dipendenze tra le trasformazioni, consentendo a Spark di ricalcolare le partizioni perse in molti scenari di errore.

Il nome racchiude tre proprietà:

• Resilient (resiliente): Spark registra la derivazione — ovvero da quale RDD precedente dipende ciascun RDD — e spesso può ricostruire le partizioni mancanti invece di mantenere repliche di ogni risultato intermedio.
• Distributed (distribuito): il dataset è rappresentato da partizioni che Spark può elaborare su executor diversi. Non è necessario disporre di più macchine fisiche: anche la modalità locale supporta gli RDD.
• Dataset (insieme di dati): un RDD può contenere coppie chiave-valore, numeri interi, stringhe, tuple e altri elementi supportati, non solo righe relazionali. Le API RDD includono operazioni come map, filter, reduceByKey, join, e aggregateByKey.

Un RDD è un’astrazione logica di raccolta e calcolo, non una dichiarazione che determinate partizioni siano assegnate in modo permanente a determinati worker.

Partizioni

Una partizione è un blocco logico di dati e normalmente corrisponde a un task all’interno di uno stage.

Supponiamo di creare 100 partizioni (queste 100 partizioni iniziali sono una scelta didattica esplicita).

Con 10.000 elementi di origine, si ottengono circa 100 elementi per partizione se l’input è distribuito in modo approssimativamente uniforme:

Figura E. Allocazione parallela delle partizioni su 100 suddivisioni iniziali dei dati

3. API RDD e API DataFrame: ottimizzazione interna di Catalyst

L’esempio originale con RDD aiuta a comprendere la lineage, le partizioni e lo shuffle.

Tuttavia, nelle tipiche operazioni ETL su dati strutturati, l’API DataFrame è il punto di partenza preferibile, perché offre colonne, informazioni sullo schema, espressioni simili a quelle SQL e operazioni visibili all’ottimizzatore.

Il diagramma seguente illustra la relazione:

Figura F. Relazione tra i concetti di grafo di Spark: lineage degli RDD, DataFrame e DAG di esecuzione di Spark

Come gli RDD hanno una lineage che definisce le trasformazioni, il piano di query di un DataFrame rappresenta la relativa query strutturata.

Spark costruisce le operazioni di esecuzione a partire dalle dipendenze e dagli operatori fisici.

L’esecuzione dei DataFrame può utilizzare internamente meccanismi basati su RDD, ma questo non significa che gli utenti debbano costruire manualmente la lineage degli RDD quando scrivono codice per i DataFrame.

3-1. Piano di query dei DataFrame

Un piano di query di un DataFrame è la sequenza di passaggi logici e fisici generata da un motore di esecuzione (riquadro viola nella Figura L).

Una query DataFrame attraversa diverse rappresentazioni:

1. Piano logico: descrizione delle operazioni richieste (scansione, filtro, raggruppamento, aggregazione).
2. Piano logico analizzato e ottimizzato: Spark risolve i riferimenti e applica le ottimizzazioni logiche pertinenti, come l’eliminazione delle colonne o il pushdown dei predicati.
3. Piano fisico: Spark sceglie gli operatori eseguibili, ad esempio scansioni, aggregazioni parziali/finali, scambi, join e ordinamenti.
4. Esecuzione: il piano fisico viene eseguito tramite i meccanismi di esecuzione di Spark e comporta, in ultima istanza, l’esecuzione di task sugli executor.

Nel nostro caso d’uso, il piano fisico può includere i seguenti passaggi:

• Aggregazione hash parziale,
• scambio per SKU e
• aggregazione hash finale.

3-2. Confronto tra le operazioni: API RDD e API DataFrame

Confrontiamo l’API RDD e l’API DataFrame usando come esempio l’aggregazione per SKU.

API RDD

Gli elementi sono (sku, qty) coppie. La nostra logica di aggregazione in Python è esplicita e Spark segue le dipendenze degli 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 DataFrame

Supponiamo che sales_df abbia le colonne sku (stringa) e qty (numerica):

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)]

In questo caso, entrambi gli esempi risolvono lo stesso problema aziendale:

Filtrare i record di vendita non validi e aggregare per SKU le quantità valide.

In entrambi i casi, collect() avvia l’esecuzione di Spark e restituisce tutti i record risultanti al Driver.

La differenza principale riguarda la rappresentazione del risultato:

• API RDD: restituisce elementi Python, come (sku, total_qty) tuple.
• API DataFrame: restituisce oggetti strutturati Row con campi denominati, come sku e total_qty.

Ecco un confronto generale tra le API RDD e DataFrame:

Tabella 2. Confronto tra funzionalità e livello di astrazione delle API RDD e DataFrame: (a cura di Kuriko IWAI)

4. Conclusioni

L’obiettivo dell’ottimizzazione di Spark non è massimizzare separatamente il numero di partizioni o l’utilizzo della CPU.

È ridurre i costi e la latenza dell’esecuzione end-to-end senza introdurre instabilità.

In questo articolo abbiamo esaminato le relazioni fondamentali del sistema:

• Risorse: i worker ospitano gli executor; gli executor eseguono i task; la configurazione di CPU e memoria limita la concorrenza.
• Programmazione RDD: le trasformazioni producono definizioni RDD immutabili e la lineage; le azioni avviano il calcolo.
• Programmazione DataFrame: le trasformazioni producono operazioni di query logiche; l’ottimizzazione sceglie gli operatori fisici eseguibili; le azioni avviano il calcolo.
• Esecuzione: Spark organizza il lavoro in job, fasi e task; i task elaborano le partizioni sugli executor. Gli shuffle e le altre dipendenze creano confini di coordinamento tra le fasi.

Comprendere come Spark esegue carichi di lavoro distribuiti è solo il punto di partenza.

Puoi scoprire un flusso di calcolo in sette passaggi usando l’API RDD.