{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 20 · Leggere il piano di esecuzione\n\nUsa 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.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pathlib import Path\nfrom tempfile import TemporaryDirectory\n\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"20-read-execution-plans\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"8\")\nspark.conf.set(\"spark.sql.adaptive.enabled\", \"true\")\nsc = spark.sparkContext\nscratch = TemporaryDirectory(prefix=\"nb20-\")\nroot = Path(scratch.name)\nsales = spark.range(100, numPartitions=2).select(\n    \"id\", (F.col(\"id\") % 5).alias(\"shop_id\"), (F.col(\"id\") % 10).alias(\"units\"),\n)\nquery = sales.filter(\"units >= 5\").groupBy(\"shop_id\").agg(F.sum(\"units\").alias(\"total\"))",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Piano di esecuzione\n\nIl piano descrive come Spark rappresenta ed esegue una query. Segui gli input dal basso verso l’output in alto.\n\n## 2. Ottimizzatore Catalyst\n\nCatalyst 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.\n\n## 3. I quattro piani\n\nL’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.\n\n## 4. Mostrare il piano di esecuzione\n\n`explain()` stampa per impostazione predefinita un piano fisico. Usa `extended` per vedere tutti e quattro i piani.\n\n## 5. Esecuzione adattiva\n\nAQE può rivedere le scelte fisiche dopo aver osservato le statistiche di esecuzione degli stage materializzati. Cerca `AdaptiveSparkPlan` attorno al lavoro idoneo.\n\n## 6. Piano formattato\n\n`explain(\"formatted\")` separa la struttura del piano fisico dai dettagli degli operatori. Abbina i nodi numerati alle sezioni di dettaglio.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "query.explain()\nquery.explain(\"extended\")\nquery.explain(\"formatted\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 7. Spark UI\n\nLa 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.\n\n## 8. Job lanciati da inferSchema\n\nLa 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.\n\n## 9. Descrizione del job\n\nImposta 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.\n\n## 10. Un job lento\n\nParti 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.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from uuid import uuid4\n\ncsv_path = root / \"sales.csv\"\ncsv_path.write_text(\"id,units\\n1,2\\n2,3\\n\", encoding=\"utf-8\")\n\ndef observe(label, operation):\n    group = uuid4().hex\n    sc.setJobGroup(group, label)\n    try:\n        result = operation()\n        print(label, sorted(sc.statusTracker().getJobIdsForGroup(group)))\n        return result\n    finally:\n        for key in (\"spark.jobGroup.id\", \"spark.job.description\", \"spark.job.interruptOnCancel\"):\n            sc.setLocalProperty(key, None)\n\nprint(\"Spark UI:\", sc.uiWebUrl)\nobserve(\"CSV with explicit schema\", lambda: spark.read.schema(\"id INT, units INT\").option(\"header\", True).csv(str(csv_path)))\nobserve(\"CSV with type inference\", lambda: spark.read.options(header=True, inferSchema=True).csv(str(csv_path)))\nobserve(\"Aggregate sales by shop\", query.collect)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 11. Spostamento dei filtri\n\nUn filtro su una chiave di raggruppamento invariata può spostarsi sotto l’aggregazione. Un filtro sul totale calcolato deve restare sopra.\n\n## 12. Leggere solo le colonne necessarie\n\nUna scansione Parquet può escludere i campi inutilizzati. Cerca lo schema di lettura della scansione.\n\n## 13. Semplificare le espressioni costanti\n\nL’ottimizzatore può semplificare espressioni composte da soli letterali. Controlla come appare `2 * 3` nel piano ottimizzato.\n\n## 14. Passare filtri al lettore di file\n\nSpark 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.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "sales.groupBy(\"shop_id\").agg(F.sum(\"units\").alias(\"total\")).filter(\n    \"shop_id = 2 AND total > 0\"\n).explain(\"extended\")\nparquet = str(root / \"sales_parquet\")\nsales.write.parquet(parquet)\nscan = spark.read.parquet(parquet).filter(\"units >= 5\").select(\n    \"units\", (F.col(\"units\") + F.expr(\"2 * 3\")).alias(\"adjusted\")\n)\nscan.explain(\"extended\")\nscan.explain(\"formatted\")\nassert scan.count() == 50",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. AQE cambia strategia di join\n\nDisattiva 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.\n\n## 16. Leggere il piano eseguito\n\nRaccogli 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.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "spark.conf.set(\"spark.sql.autoBroadcastJoinThreshold\", \"-1\")\nspark.conf.set(\"spark.sql.adaptive.autoBroadcastJoinThreshold\", \"1048576\")\nspark.conf.set(\"spark.sql.join.preferSortMergeJoin\", \"true\")\nlookup = spark.range(10, numPartitions=2).filter(\"id < 5\").select(\n    F.col(\"id\").alias(\"shop_id\"), F.concat(F.lit(\"Shop \"), F.col(\"id\")).alias(\"shop\")\n)\njoined = spark.read.parquet(parquet).join(lookup, \"shop_id\")\njoined.explain()\nassert len(joined.collect()) == 100\njoined.explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 17. Unire le piccole partizioni di shuffle con AQE\n\nAQE 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.\n\n## 18. Piano finale dopo l’azione\n\nPrima 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.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "spark.conf.set(\"spark.sql.adaptive.coalescePartitions.enabled\", \"true\")\nspark.conf.set(\"spark.sql.adaptive.coalescePartitions.parallelismFirst\", \"false\")\nsummary = spark.read.parquet(parquet).groupBy(\"shop_id\").agg(F.count(\"*\").alias(\"sales\"))\nsummary.explain()\nassert {r.shop_id: r.sales for r in summary.collect()} == {shop: 20 for shop in range(5)}\nsummary.explain()\nscratch.cleanup()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Quale piano mostra colonne e tipi risolti?\n2. Dove puoi vedere i campi richiesti a Parquet?\n3. Perché raccogliere lo stesso DataFrame prima di stampare il suo piano adattivo finale?\n\n### Risposte\n\n1. Il piano logico analizzato.\n2. Nello schema di lettura della scansione.\n3. AQE richiede statistiche di esecuzione e un’azione separata come count può usare un piano diverso.",
      "id": "cell-13"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}