Narcis Miclaus

8 min di lettura

PySpark, dalle fondamenta Lezione 31 / 60

Cos'è una partition, fisicamente

Le partition in memoria, le partition su disco, e la relazione tra partition e task.

Abbiamo usato la parola “partition” per tutto il corso. È ora di rallentare e assicurarsi che voglia dire la stessa cosa nella tua testa e nella mia, perché ci sono due sensi della parola, sono correlati ma distinti, e confonderli è la fonte di metà dei bug da partition che si vedono in giro.

Questa lezione è su cos’è una partition, fisicamente, in memoria e su disco, e qual è la sua relazione con l’unità di esecuzione che Spark fa girare davvero, ovvero un task. Una volta che quel quadro è chiaro, la prossima lezione (la knob spark.sql.shuffle.partitions e perché il default è sbagliato) avrà un senso ovvio.

La partition runtime

Una partition, nel senso runtime, è il pezzo di dati che Spark processa come singola unità di lavoro parallelo. Una partition. Un task. Un core. In ogni istante.

Questa è l’identità da interiorizzare:

1 partition -> 1 task -> 1 core, per la durata del task.

Se il tuo DataFrame ha 1000 partition e fai partire uno stage che lo processa, Spark genera 1000 task. Se il tuo cluster ha 200 core disponibili, quei 1000 task girano in 5 ondate da 200 ciascuna: in ogni momento, 200 task girano in parallelo, il resto è in coda. Se il tuo cluster ha 2000 core, tutti i 1000 task girano simultaneamente e 1000 core stanno fermi. Non puoi avere più parallelismo che partition.

Dentro un executor, una partition è semplicemente un sottoinsieme delle righe del DataFrame, tenuto in memoria (o splittato su disco se la memoria è poca). Ogni task processa la sua partition riga per riga: applica i filter, le transformation, le aggregation, qualunque cosa dica il piano dello stage, e scrive il suo output, o sui shuffle file dello stage successivo o sul sink puntato dall’action.

Le partition di un DataFrame non vivono da nessuna parte centralmente; vivono distribuite tra gli executor. È esattamente il punto del processing distribuito.

La partition su disco

L’altro uso di “partition” è su disco: una directory in una scrittura partizionata.

/data/transactions/
  year=2024/
    month=01/
      part-00000.parquet
      part-00001.parquet
      ...
    month=02/
      ...
  year=2025/
    month=01/
      ...

Quando scrivi df.write.partitionBy("year", "month").parquet(...), Spark crea una directory per ogni combinazione distinta dei valori delle colonne di partition. Rileggendo i dati, Spark usa quei nomi di directory per fare partition pruning: se filtri WHERE year = 2024, Spark apre solo i file sotto year=2024/ e salta il resto integralmente.

Queste partition su disco sono diverse dalle partition runtime. Una singola partition su disco (una directory year=2024/month=03/) può contenere molti file; leggerla produce molte partition runtime. Viceversa, una partition runtime può scrivere più partition su disco (se i dati al suo interno coprono diversi valori di year/month).

I due sensi sono correlati, entrambi parlano di spezzare i dati in pezzi, ma sono concetti separati che vivono a layer diversi. Quando qualcuno dice “questo job ha troppe partition”, chiedi quale tipo. Le partition runtime impattano sul parallelismo e sul costo dello shuffle. Le partition su disco impattano sul pruning in lettura e sul gonfiore delle directory. Le soluzioni sono diverse.

Per il resto di questa lezione, e per la maggior parte del Modulo 6, “partition” significa il senso runtime se non specificato diversamente.

Task per stage = partition dell’input

L’identità più semplice e più utile nel performance work di Spark:

Il numero di task in uno stage è uguale al numero di partition dell’input di quello stage.

Uno stage parte da un confine di shuffle (o da una read) e finisce al prossimo shuffle (o all’action). Dentro uno stage, ogni task processa una partition dell’input dello stage attraverso il piano dello stage, dall’inizio alla fine. Quindi:

  • Una read di 800 file di input (con una partition per file) crea uno stage con 800 task.
  • Un output di shuffle da groupBy con 200 partition crea uno stage a valle con 200 task.
  • Un repartition(50) su un DataFrame fa sì che lo stage successivo abbia 50 task.

Ecco perché la gente si fissa sui conteggi di partition. Il conteggio delle partition è il parallelismo. Troppo poche partition e il cluster è sotto-utilizzato; troppe e perdi tempo a schedulare task minuscoli.

Ispezionare le partition

Non devi tirare a indovinare. Due modi economici per guardare cosa sono davvero le partition di un DataFrame:

Quante partition:

df.rdd.getNumPartitions()
# 200

Questo è il conteggio runtime delle partition, quello che determina il numero di task. Funziona su qualsiasi DataFrame. Economico.

Quanto è grande ogni partition:

from pyspark.sql import functions as F

(df.withColumn("pid", F.spark_partition_id())
   .groupBy("pid")
   .count()
   .orderBy("pid")
   .show(50))

# +---+------+
# |pid| count|
# +---+------+
# |  0| 50012|
# |  1| 49873|
# |  2| 49991|
# ...
# |199| 50104|
# +---+------+

spark_partition_id() restituisce l’ID intero della partition in cui ogni riga vive in quel momento. Raggruppa per quello, conta, guarda la distribuzione. Se ogni partition ha all’incirca lo stesso conteggio di righe, i dati sono bilanciati. Se una partition ha 10x la mediana, hai skew (lezione 28) e il prossimo shuffle farà male.

Questa è la stessa diagnostica delle lezioni sullo skew, ma ora capisci esattamente cosa ti sta dicendo: il conteggio di righe per unità di lavoro parallelo. Una partition con 10 milioni di righe è un task che processa 10 milioni di righe. Una partition con 100 righe è un task che processa 100 righe. Al cluster non frega quante partition ci siano in totale; gli importa la più lenta.

Un follow-up utile: la dimensione in byte, non solo il conteggio di righe. I conteggi di righe possono essere fuorvianti: righe larghe con array e struct possono essere 100x più pesanti di righe strette di interi. Un modo grezzo per controllare, quando sospetti che lo sbilanciamento di partition sia più che semplici conteggi di righe:

def partition_size_bytes(df):
    """Per-partition byte size estimate via the RDD interface."""
    return (df.rdd
              .mapPartitionsWithIndex(
                  lambda idx, it: [(idx, sum(len(str(r)) for r in it))])
              .collect())

for pid, nbytes in sorted(partition_size_bytes(df))[:5]:
    print(f"partition {pid}: {nbytes:,} bytes (string-len approximation)")

Non è perfetto, str(row) sovrastima rispetto alla dimensione serializzata, ma i rapporti tra partition ti dicono quello che ti serve sapere. Se la partition 0 è 50x la dimensione della partition 1, hai uno skew che il solo conteggio di righe potrebbe non aver mostrato.

Conteggi di partition di default

Da dove arrivano in primo luogo i conteggi di partition? Tre fonti principali:

In lettura. Quando leggi una sorgente file-based, Parquet, CSV, JSON, ORC, Spark crea una partition per file di default, con qualche caveat:

  • Per HDFS e store HDFS-like, i file molto grandi vengono splittati per la dimensione del blocco HDFS (tipicamente 128 MB o 256 MB). Una partition per blocco.
  • Per S3 e object store simili, Spark usa spark.sql.files.maxPartitionBytes (default 128 MB) per splittare i file grossi.
  • I file minuscoli a volte vengono coalesced via spark.sql.files.openCostInBytes per evitare l’overhead di aprire molti piccoli file per nulla.

Il take-away pratico: se leggi una directory di 800 file Parquet, aspettati all’incirca 800 partition. Se quei file sono enormi, aspettatene di più. Se sono minuscoli, di meno.

Dopo uno shuffle. Ogni output di shuffle viene ripartizionato in base a spark.sql.shuffle.partitions (default 200). Group-by, join, distinct, window, qualunque cosa richieda uno shuffle, produce 200 partition dall’altro lato, indipendentemente dalla dimensione dell’input. Questo default è quasi sempre sbagliato, che è esattamente il tema della lezione 32.

Repartition esplicito. df.repartition(N) e df.coalesce(N) settano il conteggio di partition esplicitamente. Li copriremo in dettaglio nelle lezioni 33 e 34, ma per ora: repartition rishuffla a esattamente N partition, coalesce fonde le partition esistenti senza un full shuffle.

C’è anche una fonte indiretta: ogni operazione che preserva il partitioning (un filter, un select, un withColumn che non shuffla) mantiene il conteggio di partition che ha ereditato. Quindi se leggi 800 file, filtri via il 95% delle righe, e poi fai group-by, lo step di filter ha ancora 800 partition, la maggior parte delle quali minuscole, finché lo shuffle del group-by non cambia le cose. Sembra spreco, e a volte lo è, motivo per cui il Modulo 6 ha un’intera lezione su coalesce proprio per questo caso.

Una immagine mentale

Se vuoi una sola immagine da tenerti in testa, pensa alle partition come alle fette di una pizza.

  • La pizza è il tuo DataFrame.
  • Le fette sono le partition.
  • Le persone sedute attorno al tavolo sono gli executor.
  • Ogni fetta viene mangiata da una persona alla volta, un morso alla volta: quello è il task.
  • Una pizza con 200 fette e 8 persone: ogni persona mangia 25 fette, una alla volta. Il pasto è finito quando tutti hanno finito.
  • Una pizza con una fetta gigante e 199 minuscole: 199 persone finiscono le loro fette minuscole in secondi; una persona è ancora a masticare la fetta gigante 30 minuti dopo. Quello è skew.
  • Una pizza tagliata in 4 fette per 200 persone: 196 di loro non hanno niente da fare. Quello è underpartitioning.
  • Una pizza tagliata in 10000 fette microscopiche: tutti spendono più tempo a raccogliere fette che a mangiare. Quello è overpartitioning, e l’overhead è reale: ogni task ha un costo di scheduling, ogni task scrive il suo shuffle file, ogni task è un piccolo peso di bookkeeping sul driver.

Tutto il gioco è: taglia la pizza in fette circa uguali, di una dimensione che tenga ogni mangiatore occupato senza che nessuno soffochi e senza troppo overhead per fetta. La dimensione “giusta” della fetta dipende dal tuo workload, dal tuo cluster e dai tuoi dati; la prossima lezione è su come sceglierla.

Perché tutto questo conta adesso

Il Modulo 6 entrerà nelle knob e nei pattern del partitioning: il default delle shuffle partition, repartition vs. coalesce, partitioning in scrittura, bucketing. Niente di tutto questo ha senso senza il quadro di questa lezione. Prima di tunare qualunque cosa:

  • Una partition è il pezzo di dati che Spark processa come una unità.
  • Una partition = un task = un core, in ogni istante.
  • Task per stage = partition dell’input di quello stage.
  • Partition di default in lettura ~ numero di file di input (o pezzi della dimensione del blocco per file grossi).
  • Partition di default dopo lo shuffle = spark.sql.shuffle.partitions (200, quasi sempre sbagliato).
  • Le partition su disco sono un concetto separato, directory in una scrittura partizionata, anche se la parola è la stessa.

Interiorizza questo e il resto del Modulo 6 sono solo conseguenze. La lezione 32 entra dritta nel default più consequenziale di Spark e mostra come settarlo correttamente per il tuo workload.

Riferimenti: documentazione Apache Spark sul partitioning RDD e sul modello di esecuzione SQL; guida al tuning di Databricks su partition e parallelismo. Recuperati il 2026-05-01.

Cerca