20 · Leggere il piano di esecuzione

Usa una piccola query sulle vendite per collegare gli operatori del piano al lavoro di Spark. Esegui in una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa con %pip install "pyspark>=4.0,<4.1", poi usa Run All.

from pathlib import Path
from tempfile import TemporaryDirectory

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

spark = SparkSession.builder.master("local[2]").appName("20-read-execution-plans").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "8")
spark.conf.set("spark.sql.adaptive.enabled", "true")
sc = spark.sparkContext
scratch = TemporaryDirectory(prefix="nb20-")
root = Path(scratch.name)
sales = spark.range(100, numPartitions=2).select(
    "id", (F.col("id") % 5).alias("shop_id"), (F.col("id") % 10).alias("units"),
)
query = sales.filter("units >= 5").groupBy("shop_id").agg(F.sum("units").alias("total"))

1. Piano di esecuzione

Il piano descrive come Spark rappresenta ed esegue una query. Segui gli input dal basso verso l’output in alto.

2. Ottimizzatore Catalyst

Catalyst risolve e riscrive le espressioni logiche e contribuisce a scegliere gli operatori fisici. Un filtro o una proiezione nel codice può spostarsi o sparire durante l’ottimizzazione.

3. I quattro piani

L’output esteso mostra il piano logico dopo il parsing, quello analizzato, quello ottimizzato e il piano fisico. Il piano analizzato risolve nomi e tipi. Quello fisico sceglie operatori eseguibili.

4. Mostrare il piano di esecuzione

explain() stampa per impostazione predefinita un piano fisico. Usa extended per vedere tutti e quattro i piani.

5. Esecuzione adattiva

AQE può rivedere le scelte fisiche dopo aver osservato le statistiche di esecuzione degli stage materializzati. Cerca AdaptiveSparkPlan attorno al lavoro idoneo.

6. Piano formattato

explain("formatted") separa la struttura del piano fisico dai dettagli degli operatori. Abbina i nodi numerati alle sezioni di dettaglio.

query.explain()
query.explain("extended")
query.explain("formatted")

7. Spark UI

La vista SQL collega i piani alle metriche di esecuzione. Le viste Jobs e Stages mostrano l’attività dei task e lo shuffle. Un notebook ospitato può limitare l’accesso alla porta locale.

8. Job lanciati da inferSchema

La deduzione dei tipi CSV può avviare lavoro mentre si crea il risultato della lettura, prima di un’azione esplicita sul DataFrame. Confronta le due letture con il tracker di stato.

9. Descrizione del job

Imposta una descrizione prima di inviare il lavoro per ritrovarlo più facilmente. L’esempio usa gruppi di job per etichettarlo e osservarlo, poi cancella le proprietà locali del thread.

10. Un job lento

Parti dallo stage costoso ed esamina durata dei task, shuffle e spill. Lo sbilanciamento concentra il lavoro in poche partizioni. Lo spill scrive su disco dati di esecuzione sotto pressione di memoria. L’esempio è troppo piccolo per simulare queste condizioni.

from uuid import uuid4

csv_path = root / "sales.csv"
csv_path.write_text("id,units\n1,2\n2,3\n", encoding="utf-8")

def observe(label, operation):
    group = uuid4().hex
    sc.setJobGroup(group, label)
    try:
        result = operation()
        print(label, sorted(sc.statusTracker().getJobIdsForGroup(group)))
        return result
    finally:
        for key in ("spark.jobGroup.id", "spark.job.description", "spark.job.interruptOnCancel"):
            sc.setLocalProperty(key, None)

print("Spark UI:", sc.uiWebUrl)
observe("CSV with explicit schema", lambda: spark.read.schema("id INT, units INT").option("header", True).csv(str(csv_path)))
observe("CSV with type inference", lambda: spark.read.options(header=True, inferSchema=True).csv(str(csv_path)))
observe("Aggregate sales by shop", query.collect)

11. Spostamento dei filtri

Un filtro su una chiave di raggruppamento invariata può spostarsi sotto l’aggregazione. Un filtro sul totale calcolato deve restare sopra.

12. Leggere solo le colonne necessarie

Una scansione Parquet può escludere i campi inutilizzati. Cerca lo schema di lettura della scansione.

13. Semplificare le espressioni costanti

L’ottimizzatore può semplificare espressioni composte da soli letterali. Controlla come appare 2 * 3 nel piano ottimizzato.

14. Passare filtri al lettore di file

Spark può passare predicati idonei al lettore. Cerca PushedFilters nei dettagli formattati della scansione. Un predicato passato può aiutare a saltare dati. Spark può comunque mantenere un filtro per garantire il risultato.

sales.groupBy("shop_id").agg(F.sum("units").alias("total")).filter(
    "shop_id = 2 AND total > 0"
).explain("extended")
parquet = str(root / "sales_parquet")
sales.write.parquet(parquet)
scan = spark.read.parquet(parquet).filter("units >= 5").select(
    "units", (F.col("units") + F.expr("2 * 3")).alias("adjusted")
)
scan.explain("extended")
scan.explain("formatted")
assert scan.count() == 50

15. AQE cambia strategia di join

Disattiva il broadcast automatico iniziale e consenti quello adattivo. La query parte da un join per uguaglianza idoneo e da una piccola tabella di riferimento filtrata. AQE può passare da sort-merge a broadcast hash una volta note le dimensioni di esecuzione. Controlla il piano effettivo perché la scelta dipende da statistiche e configurazione.

16. Leggere il piano eseguito

Raccogli esattamente il DataFrame che stai esaminando. Chiamare count può eseguire una query diversa. Stampa di nuovo il piano fisico e cerca piano finale e iniziale quando AQE li fornisce entrambi.

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "1048576")
spark.conf.set("spark.sql.join.preferSortMergeJoin", "true")
lookup = spark.range(10, numPartitions=2).filter("id < 5").select(
    F.col("id").alias("shop_id"), F.concat(F.lit("Shop "), F.col("id")).alias("shop")
)
joined = spark.read.parquet(parquet).join(lookup, "shop_id")
joined.explain()
assert len(joined.collect()) == 100
joined.explain()

17. Unire le piccole partizioni di shuffle con AQE

AQE può combinare partizioni piccole dopo lo shuffle in base alle statistiche dell’output map. Cerca una lettura adattiva di shuffle con partizioni accorpate. Questo è diverso dal cambiare manualmente il numero di partizioni di input.

18. Piano finale dopo l’azione

Prima dell’esecuzione Spark non conosce le scelte adattive finali. Confronta la stessa query prima e dopo aver raccolto i risultati. Verifica i valori di output indipendentemente dalle etichette del piano.

spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")
summary = spark.read.parquet(parquet).groupBy("shop_id").agg(F.count("*").alias("sales"))
summary.explain()
assert {r.shop_id: r.sales for r in summary.collect()} == {shop: 20 for shop in range(5)}
summary.explain()
scratch.cleanup()

Verifica la comprensione

1. Quale piano mostra colonne e tipi risolti?
2. Dove puoi vedere i campi richiesti a Parquet?
3. Perché raccogliere lo stesso DataFrame prima di stampare il suo piano adattivo finale?

Risposte

1. Il piano logico analizzato.
2. Nello schema di lettura della scansione.
3. AQE richiede statistiche di esecuzione e un’azione separata come count può usare un piano diverso.