{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 04 · Capire come Spark esegue\n\nSegui 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**.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from tempfile import TemporaryDirectory\nfrom uuid import uuid4\n\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = (\n    SparkSession.builder.master(\"local[2]\")\n    .appName(\"04-understand-spark-execution\")\n    .getOrCreate()\n)\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nsc = spark.sparkContext\ntracker = sc.statusTracker()\n\nsales = spark.range(12, numPartitions=2).select(\n    F.col(\"id\").alias(\"sale_id\"),\n    (F.col(\"id\") % 3).alias(\"shop_id\"),\n    (F.col(\"id\") + 1).alias(\"units\"),\n)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. DataFrame\n\nUn 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.\n\n## 2. Partizione\n\nUna 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.\n\n## 3. Trasformazione\n\nUna trasformazione descrive un nuovo DataFrame a partire da uno esistente. L’esempio filtra le righe e usa `withColumn()` per aggiungere un campo derivato.\n\n## 4. Trasformazione stretta\n\nCon questi filtri e proiezioni, ogni partizione di output dipende soltanto dalla corrispondente partizione di input. Spark può elaborarla senza redistribuire le righe.\n\n## 5. Trasformazione larga\n\nUn’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.\n\n## 6. Lineage\n\nIl 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.\n\n## 7. Immutabilità del DataFrame\n\nOgni trasformazione restituisce un altro DataFrame. L’originale conserva colonne e significato.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "print(\"Source partitions:\", sales.rdd.getNumPartitions())\nprepared = sales.filter(F.col(\"units\") >= 3).withColumn(\"adjusted_units\", F.col(\"units\") * 2)\ntotals = prepared.groupBy(\"shop_id\").agg(F.sum(\"adjusted_units\").alias(\"total\"))\nassert \"adjusted_units\" not in sales.columns\nassert \"adjusted_units\" in prepared.columns",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 8. Azione\n\nUn’azione richiede un risultato o un effetto. Qui `collect()` restituisce a Python i totali dei tre negozi.\n\n## 9. Job\n\nUn 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.\n\n## 10. Valutazione pigra\n\nDi 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.\n\n## 11. Contare i job avviati\n\nUsa 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.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def observe_jobs(label, operation):\n    group = f\"nb04-{uuid4()}\"\n    sc.setJobGroup(group, label)\n    try:\n        result = operation()\n        ids = sorted(tracker.getJobIdsForGroup(group))\n        print(f\"{label}: {len(ids)} tracked job(s), IDs {ids}\")\n        return result, ids\n    finally:\n        for key in (\"spark.jobGroup.id\", \"spark.job.description\", \"spark.job.interruptOnCancel\"):\n            sc.setLocalProperty(key, None)\n\nquery, _ = observe_jobs(\"Build query\", lambda: totals.filter(F.col(\"total\") > 0))\nrows, action_ids = observe_jobs(\"Collect shop totals\", query.collect)\nassert {row.shop_id: row.total for row in rows} == {0: 42, 1: 48, 2: 60}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 12. Azione dentro un ciclo\n\nContare 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.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def count_each_shop():\n    return {shop: prepared.filter(F.col(\"shop_id\") == shop).count() for shop in range(3)}\n\nloop_counts, _ = observe_jobs(\"Count each shop separately\", count_each_shop)\ngroup_counts, _ = observe_jobs(\"Collect grouped counts\", prepared.groupBy(\"shop_id\").count().collect)\nassert loop_counts == {row.shop_id: row[\"count\"] for row in group_counts}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 13. Rimescolamento\n\nUn 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.\n\n## 14. Stage\n\nUno 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.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "totals.explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Driver\n\nIl driver coordina l’applicazione, prepara l’esecuzione e pianifica il lavoro. Il kernel del notebook è la parte Python del driver in questo esempio.\n\n## 16. Executor\n\nUn 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.\n\n## 17. Task\n\nUn task è un’unità di lavoro su una partizione all’interno di uno stage. Se ci sono slot disponibili, più task possono essere eseguiti contemporaneamente.\n\n## 18. Esecuzione locale\n\nCon `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.\n\n## 19. Gestore del cluster\n\nUn gestore del cluster assegna risorse alle applicazioni Spark. Tra gli esempi ci sono Spark Standalone, YARN e Kubernetes.\n\n## 20. Rimozione di Mesos in Spark 4.0\n\nI materiali meno recenti possono citare Mesos come gestore del cluster. Spark 4.0 ha rimosso il supporto a Mesos.\n\n## 21. Gerarchia job, stage e task\n\nUn 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.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "settings = sc.getConf()\nprint(\"Master:\", settings.get(\"spark.master\"))\nprint(\"SQL shuffle partitions:\", spark.conf.get(\"spark.sql.shuffle.partitions\"))\n\nfor job_id in action_ids:\n    job = tracker.getJobInfo(job_id)\n    if job is not None:\n        for stage_id in job.stageIds:\n            stage = tracker.getStageInfo(stage_id)\n            if stage is not None:\n                print(f\"Job {job_id}, stage {stage_id}, tasks {stage.numTasks}\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 22. DAG\n\nUn grafo orientato aciclico rappresenta le dipendenze. La direzione indica quale lavoro dipende da quello precedente. Il grafo è privo di cicli di dipendenza.\n\n## 23. DAG scheduler e task scheduler\n\nIl 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.\n\n## 24. Stage saltato\n\nSpark 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.\n\n## 25. Mostrare il grafo delle dipendenze\n\nSe 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.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "print(\"Spark UI:\", sc.uiWebUrl)\ndebug_view = totals.rdd.toDebugString()\nprint(debug_view.decode(\"utf-8\") if isinstance(debug_view, bytes) else debug_view)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 26. Tolleranza ai guasti\n\nSpark 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.\n\n## 27. Salvare un checkpoint\n\nUn 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.\n\n## 28. Checkpoint e cache\n\nLa 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.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "cached = totals.cache()\ntry:\n    assert cached.count() == 3\n    print(\"Cached:\", cached.is_cached)\nfinally:\n    cached.unpersist()\n\nwith TemporaryDirectory(prefix=\"spark-nb04-\") as checkpoint_dir:\n    sc.setCheckpointDir(checkpoint_dir)\n    checkpointed = totals.checkpoint(eager=True)\n    checkpointed.explain()\n    assert {row.shop_id: row.total for row in checkpointed.collect()} == {\n        row.shop_id: row.total for row in rows\n    }",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 29. Esecuzione speculativa\n\nQuando è 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.",
      "id": "cell-17"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "print(\"Speculation enabled:\", sc.getConf().get(\"spark.speculation\", \"false\"))",
      "execution_count": null,
      "outputs": [],
      "id": "cell-18"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché il filtro può essere eseguito all’interno di ogni partizione di origine?\n2. Perché raggruppare per negozio può richiedere uno scambio?\n3. Come cambia il checkpoint le dipendenze del DataFrame restituito?\n\n### Risposte\n\n1. Il filtro può decidere se mantenere una riga usando i valori già presenti in quella riga.\n2. Le vendite di un negozio possono trovarsi in più partizioni. I totali parziali devono confluire prima che Spark calcoli quello finale.\n3. Il DataFrame con checkpoint legge il risultato materializzato e può quindi fare a meno di rieseguire il piano a monte.",
      "id": "cell-19"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}