{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 21 · Gestire partizioni, cache e memoria\n\nMisura come si distribuisce il lavoro, prova una tecnica contro lo sbilanciamento e osserva la cache. Usa una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa PySpark con `%pip install \"pyspark>=4.0,<4.1\"`. Tutti i dati sono generati localmente.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pathlib import Path\nfrom tempfile import TemporaryDirectory\n\nfrom pyspark import StorageLevel\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"21-partitions-cache-memory\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"4\")\nspark.conf.set(\"spark.sql.adaptive.enabled\", \"false\")\nsales = spark.range(1000, numPartitions=4).select(\n    \"id\",\n    F.when(F.col(\"id\") < 900, 0).otherwise(F.col(\"id\") % 7 + 1).alias(\"shop_id\"),\n    (F.col(\"id\") % 10 + 1).alias(\"units\"),\n)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Contare le partizioni\n\nIn PySpark classico, `df.rdd.getNumPartitions()` indica quante partizioni ha la rappresentazione RDD del DataFrame. Il conteggio da solo non indica come si distribuiscono le righe.\n\n## 2. Cache\n\nMetti in cache un risultato quando più azioni lo riuseranno e il costo di memorizzazione è giustificato. Usa quel DataFrame per le azioni successive e libera la cache quando hai finito.\n\n## 3. Sbilanciamento\n\nIl lavoro è sbilanciato quando si distribuisce in modo disomogeneo. Dopo aver ripartito per negozio, confronta la partizione più grande con le altre. Il numero di righe è una stima del lavoro. I costi effettivi dei task possono variare.\n\n## 4. Cambiare il numero di partizioni\n\n`repartition()` richiede uno shuffle e può aumentare o ridurre il numero di partizioni. `coalesce()` può ridurlo senza introdurre uno shuffle, ma può lasciare il lavoro sbilanciato.\n\n## 5. Chiavi che raccolgono molte righe\n\nIl partizionamento per hash mantiene insieme le chiavi uguali. Il negozio 0 ha 900 righe, quindi almeno 900 finiscono nella stessa partizione anche aumentando il numero totale di partizioni.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "print(\"Input partitions:\", sales.rdd.getNumPartitions())\ncached_sales = sales.cache()\ntry:\n    assert cached_sales.count() == 1000\n    assert cached_sales.filter(\"shop_id = 0\").count() == 900\nfinally:\n    cached_sales.unpersist()\n\nkeyed = sales.repartition(4, \"shop_id\")\nassert keyed.rdd.getNumPartitions() == 4\ncounts = keyed.select(F.spark_partition_id().alias(\"partition\")).groupBy(\"partition\").count()\nassert max(row[\"count\"] for row in counts.collect()) >= 900\ncounts.orderBy(\"partition\").show()\nassert sales.coalesce(2).rdd.getNumPartitions() == 2",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 6. Dividere le partizioni sbilanciate con AQE\n\nL’esecuzione adattiva può suddividere partizioni di shuffle sbilanciate per i join idonei e replicare il lavoro corrispondente quando necessario. Usa dimensioni osservate e soglie. Controlla il piano eseguito e le metriche dei task. Qui lo sbilanciamento si vede dai conteggi, anche senza una suddivisione da parte di AQE.\n\n## 7. Aggiungere sale a una chiave frequente\n\nAssegna le righe della chiave frequente a più gruppi intermedi. Qui un hash stabile dell’identificativo rende riproducibile l’assegnazione. Il sale suddivide una chiave logica.\n\n## 8. Aggregazione parziale\n\nSomma in ogni gruppo con sale, poi somma i risultati parziali per chiave originale. Spark può già calcolare somme parziali per le somme native. Qui il sale dimostra la correttezza del risultato senza misurare un guadagno di prestazioni.\n\n## 9. Aggregazione senza un risultato parziale combinabile\n\nUna funzione che richiede tutti i valori di un gruppo può doverli conservare. Per esempio, in generale le mediane dei gruppi con sale non bastano a ricavare la mediana globale. Usa un percentile integrato esatto o approssimato quando è adatto.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "salt = F.when(F.col(\"shop_id\") == 0, F.pmod(F.xxhash64(\"id\"), F.lit(4))).otherwise(0)\npartial = sales.withColumn(\"salt\", salt).groupBy(\"shop_id\", \"salt\").agg(F.sum(\"units\").alias(\"units\"))\ncombined = partial.groupBy(\"shop_id\").agg(F.sum(\"units\").alias(\"units\"))\ndirect = sales.groupBy(\"shop_id\").agg(F.sum(\"units\").alias(\"units\"))\nassert {tuple(row) for row in combined.collect()} == {tuple(row) for row in direct.collect()}\nassert partial.filter(\"shop_id = 0\").count() == 4\npartial.filter(\"shop_id = 0\").orderBy(\"salt\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 10. Riversamento su disco\n\nGli operatori possono scrivere dati intermedi su disco quando la memoria scarseggia. Controlla le metriche di spill in memoria e su disco per task. Un task lento con spill elevato può avere una partizione di input grande. L’esempio evita di forzare la pressione sulla memoria.\n\n## 11. Dimensione delle partizioni in lettura\n\n`spark.sql.files.maxPartitionBytes` influisce su come Spark raggruppa l’input suddivisibile in partizioni di scansione. Contano anche formato, dimensioni e costo di apertura dei file.\n\n## 12. Partizioni dopo lo shuffle\n\n`spark.sql.shuffle.partitions` imposta il numero predefinito per molti shuffle SQL. AQE può cambiare come viene raggruppato il lavoro successivo. Qui è disabilitata per rendere riproducibile il conteggio.\n\n## 13. coalesce(1) prima di un calcolo pesante\n\nRidurre a una partizione può far eseguire il lavoro successivo a un solo task, fino a un altro scambio. Svolgi prima il lavoro in parallelo se alla fine serve davvero un unico piccolo file.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "with TemporaryDirectory(prefix=\"nb21-\") as directory:\n    path = str(Path(directory) / \"sales\")\n    sales.write.parquet(path)\n    file_scan = spark.read.parquet(path)\n    print(\"Max scan partition bytes:\", spark.conf.get(\"spark.sql.files.maxPartitionBytes\"))\n    print(\"Scan partitions:\", file_scan.rdd.getNumPartitions())\n\nprint(\"Configured shuffle partitions:\", spark.conf.get(\"spark.sql.shuffle.partitions\"))\nassert direct.rdd.getNumPartitions() == 4\nsales.coalesce(1).select(F.sqrt(\"units\")).explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 14. Livello di memorizzazione\n\n`persist()` permette di scegliere un livello di memorizzazione. Qui `MEMORY_AND_DISK_DESER` conserva i blocchi in memoria quando possibile e consente l’uso del disco.\n\n## 15. Liberare la cache\n\n`unpersist()` rimuove la richiesta di persistenza e libera i blocchi. Usa `blocking=True` per attendere la rimozione.\n\n## 16. Controllare se un DataFrame è in cache\n\n`is_cached` mostra l’indicatore di persistenza dell’oggetto DataFrame. Il numero di partizioni già calcolate va verificato separatamente.\n\n## 17. Cache pigra\n\nLa persistenza contrassegna un piano per la cache. Un’azione la popola man mano che valuta le partizioni. Se accessibile, la vista Storage dell’interfaccia Spark mostra i blocchi materializzati.\n\n## 18. Cache per un DataFrame usato una volta\n\nLa prima azione deve calcolare i dati e popolare la cache. Un risultato usato una sola volta spesso sostiene il costo di memorizzazione senza beneficiare di una seconda lettura più veloce.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "reused = sales.filter(\"units >= 5\").persist(StorageLevel.MEMORY_AND_DISK_DESER)\ntry:\n    assert reused.is_cached\n    print(\"Requested storage level:\", reused.storageLevel)\n    print(\"Spark UI:\", spark.sparkContext.uiWebUrl)\n    assert reused.count() == 600\n    assert reused.agg(F.max(\"units\")).first()[0] == 10\nfinally:\n    reused.unpersist(blocking=True)\nassert not reused.is_cached",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 19. Ridurre lo shuffle\n\nFiltra o escludi colonne superflue prima di uno scambio se la semantica lo permette. Aggregare alla grana richiesta può ridurre le righe spostate.\n\n## 20. Riuso del partizionamento\n\nSpark può riusare una distribuzione compatibile con i requisiti dell’operatore a valle. Qui l’aggregazione può usare il partizionamento per hash esplicito sul negozio. Cerca gli scambi nel piano per verificare se due operazioni sulla stessa chiave condividono la distribuzione.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "prepared = sales.filter(\"units >= 5\").select(\"shop_id\", \"units\").repartition(4, \"shop_id\")\nprepared.groupBy(\"shop_id\").agg(F.sum(\"units\")).explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 21. Memoria riservata\n\nNella configurazione standard Spark calcola la memoria unificata sottraendo una riserva di 300 MiB dall’heap JVM. L’overhead del container è una risorsa separata.\n\n## 22. Memoria unificata\n\nEsecuzione e memorizzazione condividono una regione dimensionata da `spark.memory.fraction`. L’esecuzione include shuffle, join e ordinamenti. La memorizzazione include i blocchi in cache.\n\n## 23. Memoria protetta per la cache\n\n`spark.memory.storageFraction` imposta la quota di memoria unificata sotto la quale l’esecuzione non può espellere blocchi. La quota rappresenta una soglia di protezione, non una preallocazione esclusiva.\n\n## 24. Memoria del driver\n\nIl driver richiede spazio per pianificazione, metadati e risultati restituiti. L’heap JVM e la memoria del processo Python sono separati. Configura l’heap del driver all’avvio.\n\n## 25. Memoria esaurita nel driver\n\n`collect()` o `toPandas()` senza limiti possono sovraccaricare il driver. Controlla i log e usa anteprime limitate o scritture distribuite.\n\n## 26. Memoria esaurita negli executor\n\nUna grande partizione di shuffle, un broadcast, un risultato in cache o un worker Python possono mettere sotto pressione le risorse. Individua il processo o container fallito, controlla sbilanciamento e metriche dei task e adegua la disposizione dei dati o le risorse. Gli esempi evitano di provocare un esaurimento di memoria.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "conf = spark.sparkContext.getConf()\nprint(\"Memory fraction:\", conf.get(\"spark.memory.fraction\", \"0.6\"))\nprint(\"Storage fraction:\", conf.get(\"spark.memory.storageFraction\", \"0.5\"))\nprint(\"Bounded preview:\", sales.orderBy(\"id\").take(3))",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché più partizioni possono lasciare sovraccarico un negozio?\n2. Perché la somma delle somme parziali con sale recupera il totale?\n3. Che cosa indica is_cached prima della prima azione?\n\n### Risposte\n\n1. Il partizionamento per hash manda comunque chiavi uguali nella stessa partizione.\n2. Ogni riga contribuisce a una sola somma parziale e la somma finale le combina tutte.\n3. Indica la richiesta di persistenza. La cache deve ancora essere materializzata da un’azione.",
      "id": "cell-15"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}