{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 10 · Aggregare e cambiare forma\n\nRiepiloga le vendite del bar alla grana scelta e cambia la forma del risultato. La lezione autonoma usa PySpark classico 4.0.x e Java 17 o successivo. In una nuova sessione CPU su Kaggle, installa le dipendenze mancanti con `%pip install \"pyspark[sql]>=4.0,<4.1\"` in una cella di preparazione, poi usa Run All. L’extra SQL fornisce pandas e PyArrow per l’aggregazione personalizzata.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "import pandas as pd\nfrom pyspark.errors import AnalysisException\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"10-aggregate-and-reshape\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nsales = spark.createDataFrame(\n    [(1, \"North\", \"tea\", 2, 500),\n     (2, \"North\", \"tea\", 1, 250),\n     (3, \"North\", \"cake\", 1, None),\n     (4, \"South\", \"tea\", 3, 750),\n     (5, None, \"cake\", 2, 700)],\n    \"line_id INT, shop STRING, item STRING, units INT, revenue_cents INT\",\n)\nsales.orderBy(\"line_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Raggruppare le righe\n\n`groupBy(\"shop\")` definisce i gruppi per negozio. Anche le righe con negozio null formano un gruppo.\n\n## 2. Calcolare aggregati per gruppo\n\n`agg()` calcola una riga di risultato per gruppo. Qui la somma conta le unità alla grana del negozio.\n\n## 3. Contare i valori distinti\n\n`count_distinct(\"item\")` conta i nomi di prodotto distinti e non null per negozio.\n\n## 4. Funzioni di aggregazione\n\nFunzioni come `sum()`, `avg()`, `min()` e `max()` riassumono più righe in un valore per gruppo. Un alias assegna un nome a ogni risultato.\n\n## 5. Contare righe e valori presenti\n\n`count(\"*\")` conta tutte le righe. `count(\"revenue_cents\")` conta quelle con un ricavo presente.\n\n## 6. Conteggio distinto approssimato\n\n`approx_count_distinct()` stima la cardinalità con una struttura compatta (*sketch*). Il parametro della deviazione standard relativa controlla l’accuratezza richiesta. Usa un conteggio esatto per le asserzioni sui piccoli dati di prova.\n\n## 7. count(colonna) ignora i null\n\nNorth ha tre righe e solo due valori di ricavo registrati. Usa entrambi i conteggi per mostrare la copertura.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "by_shop = sales.groupBy(\"shop\").agg(\n    F.sum(\"units\").alias(\"units_sold\"),\n    F.count_distinct(\"item\").alias(\"exact_items\"),\n    F.avg(\"units\").alias(\"mean_units\"),\n    F.count(\"*\").alias(\"lines\"),\n    F.count(\"revenue_cents\").alias(\"priced_lines\"),\n    F.approx_count_distinct(\"item\").alias(\"estimated_items\"),\n)\nby_shop.orderBy(F.col(\"shop\").asc_nulls_last()).show()\nnorth = by_shop.filter(\"shop = 'North'\").first()\nassert (north.units_sold, north.exact_items, north.lines, north.priced_lines) == (4, 2, 3, 2)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 8. Mediana e percentili\n\nUsa `median()` per la mediana e `percentile_approx()` per quantili approssimati. Un percentile approssimato può scegliere un valore osservato laddove la mediana esatta interpolerebbe.\n\n## 9. Raccogliere i valori in una lista\n\n`collect_list()` conserva i duplicati tra i valori presenti. L’ordine dopo uno shuffle è indefinito. Ordina l’array risultante per una visualizzazione deterministica. Gruppi grandi possono produrre array grandi.\n\n## 10. Aggregazione globale\n\n`sales.agg()` aggrega l’intero DataFrame e restituisce una riga.\n\n## 11. Raggruppare per un’espressione\n\nUna chiave di gruppo può essere un’espressione. Qui `units >= 2` divide le righe in gruppi da una unità e da più unità.\n\n## 12. Aggregazione condizionale\n\nCombina una condizione e un aggregato per misurare un sottoinsieme accanto al gruppo intero. Un valore predefinito pari a zero fa contribuire con zero unità ogni riga che non soddisfa la condizione.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "sales.agg(\n    F.median(\"units\").alias(\"median_units\"),\n    F.percentile_approx(\"units\", [0.25, 0.75]).alias(\"quartiles\"),\n    F.sort_array(F.collect_list(\"item\")).alias(\"items\"),\n).show(truncate=False)\nsales.groupBy((F.col(\"units\") >= 2).alias(\"multi_unit\")).count().show()\ntea_units = F.sum(F.when(F.col(\"item\") == \"tea\", F.col(\"units\")).otherwise(0))\nassert sales.agg(tea_units).first()[0] == 6\nsales.groupBy(\"shop\").agg(tea_units.alias(\"tea_units\")).show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 13. Subtotali\n\n`rollup(\"shop\", \"item\")` aggiunge ai gruppi negozio-prodotto i subtotali per negozio e il totale generale.\n\n## 14. Null nelle righe di subtotale\n\nSpark rappresenta con null una dimensione esclusa dal raggruppamento del subtotale. Anche una riga di dettaglio può avere un vero null per il negozio.\n\n## 15. Riconoscere le righe di subtotale\n\n`grouping(column)` restituisce 1 se la colonna è esclusa dal raggruppamento e 0 se vi partecipa. Usa questi indicatori per distinguere i marcatori di subtotale dai null nei dati.\n\n## 16. Tutte le combinazioni di subtotali\n\n`cube()` include ogni sottoinsieme delle dimensioni di raggruppamento. Con negozio e prodotto aggiunge i subtotali per solo prodotto alle combinazioni di rollup.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def summarize(grouped):\n    return grouped.agg(\n        F.sum(\"units\").alias(\"units_sold\"),\n        F.grouping(\"shop\").alias(\"all_shops\"),\n        F.grouping(\"item\").alias(\"all_items\"),\n    )\n\nrolled = summarize(sales.rollup(\"shop\", \"item\"))\ncubed = summarize(sales.cube(\"shop\", \"item\"))\nrolled.orderBy(\"all_shops\", \"all_items\", \"shop\", \"item\").show()\nassert rolled.filter(\"all_shops = 1 AND all_items = 1\").first().units_sold == 9\nassert rolled.filter(\"shop IS NULL AND all_shops = 0 AND all_items = 0\").count() == 1\nassert cubed.filter(\"all_shops = 1 AND all_items = 0\").count() == 2",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 17. Aggregazione personalizzata\n\nUna pandas UDF di aggregazione per gruppo riceve i valori di un gruppo come Series pandas e restituisce uno scalare. Qui calcola la quota di righe con almeno due unità.\n\n## 18. Gruppo intero nella memoria dell’executor\n\nUna pandas UDF di aggregazione carica in memoria tutti i valori del gruppo. Un gruppo sbilanciato può esaurire la memoria anche se gli altri sono piccoli. Le aggregazioni integrate possono spesso combinare risultati parziali.\n\n## 19. Calcolare un indicatore per gruppo\n\nEsprimi la metrica come media di un indicatore che vale 0 o 1. L’esempio confronta il risultato integrato con l’aggregazione personalizzata sugli stessi dati senza null.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "@F.pandas_udf(\"double\")\ndef multi_share(values: pd.Series) -> float:\n    return float((values >= 2).mean())\n\ncustom = sales.groupBy(\"shop\").agg(multi_share(\"units\").alias(\"share\"))\nnative = sales.groupBy(\"shop\").agg(F.avg((F.col(\"units\") >= 2).cast(\"int\")).alias(\"share\"))\nassert {r.shop: r.share for r in custom.collect()} == {r.shop: r.share for r in native.collect()}\nnative.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 20. Pivot\n\n`pivot()` sposta le categorie di prodotto nelle colonne. Raggruppa prima per negozio e scegli la misura da aggregare in ogni cella.\n\n## 21. Pivot senza elenco di valori\n\nSenza un elenco esplicito, Spark cerca subito i valori distinti del pivot per determinare lo schema di output. Usa un elenco quando conosci le categorie previste.\n\n## 22. count(\"*\") dentro un pivot\n\n`count(\"*\")` conta le righe di input di una cella, incluse quelle con ricavo sconosciuto. `count(\"revenue_cents\")` conta i ricavi presenti. Per la torta, North ha una riga e zero ricavi noti.\n\n## 23. Celle vuote dopo il pivot\n\nUna coppia negozio-prodotto senza righe può produrre una cella null. North ha una riga per la torta con ricavo sconosciuto. South non ha righe per la torta. La somma dei ricavi può essere null in entrambi i casi. Usa il conteggio delle righe per distinguerli.\n\n## 24. Più misure nel pivot\n\nPassa più aggregati con alias ad `agg()` per ottenere una colonna per categoria e misura.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "items = [\"tea\", \"cake\"]\nwide = sales.groupBy(\"shop\").pivot(\"item\", items).sum(\"units\")\nwide.orderBy(\"shop\").show()\ndiscovered = sales.groupBy(\"shop\").pivot(\"item\").sum(\"units\")\nassert set(discovered.columns) == {\"shop\", \"tea\", \"cake\"}\n\nmetrics = sales.groupBy(\"shop\").pivot(\"item\", items).agg(\n    F.count(\"*\").alias(\"lines\"),\n    F.count(\"revenue_cents\").alias(\"priced\"),\n    F.sum(\"revenue_cents\").alias(\"revenue\"),\n)\nnorth_metrics = metrics.filter(\"shop = 'North'\").first()\nassert (north_metrics.cake_lines, north_metrics.cake_priced, north_metrics.cake_revenue) == (1, 0, None)\nassert wide.filter(\"shop = 'South'\").first().cake is None\nmetrics.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 25. fillna su più colonne\n\nUna chiamata scalare a `fillna(0)` può riempire le colonne numeriche compatibili in tutto il risultato. Limita le sostituzioni alle misure per cui zero ha un significato definito. Qui i conteggi di unità assenti diventano zero. I ricavi sconosciuti restano tali.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "wide_units = wide.fillna(0, subset=items)\nassert wide_units.filter(\"shop = 'South'\").first().cake == 0\nwide_units.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 26. Unpivot\n\n`unpivot()` trasforma colonne di misure in coppie etichetta-valore. Mantieni le colonne identificative e assegna un nome alle nuove colonne per categoria e valore.\n\n## 27. Unpivot con stack\n\n`stack()` può esprimere la stessa struttura specificando ogni etichetta e il suo valore. `selectExpr()` accetta espressioni SQL come stringhe. Usa apici singoli per le etichette letterali e identificativi di colonna per i rispettivi valori.\n\n## 28. Tipi diversi nell’unpivot\n\nLe colonne dei valori devono avere un tipo comune. Spark può promuovere tipi compatibili come INT e BIGINT. Qui una stringa e un numero fanno fallire l’analisi. Converti esplicitamente i valori quando la conversione ne rispetta il significato.\n\n## 29. Pivot e unpivot non sono sempre inversi\n\nIl pivot aggrega le due righe del tè di North in una cella con tre unità. L’unpivot recupera il riepilogo negozio-prodotto. Gli ID delle righe originali e le quantità separate si perdono nell’aggregazione. Riempire le celle vuote introduce anche coppie con valore zero esplicito.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "long_units = wide_units.unpivot(\"shop\", items, \"item\", \"units\")\nstacked = wide_units.selectExpr(\"shop\", \"stack(2, 'tea', tea, 'cake', cake) AS (item, units)\")\nassert {(r.shop, r.item, r.units) for r in long_units.collect()} == {\n    (r.shop, r.item, r.units) for r in stacked.collect()\n}\nlong_units.orderBy(\"shop\", \"item\").show()\n\nmixed = wide_units.select(\"shop\", \"tea\", F.col(\"cake\").cast(\"string\").alias(\"cake_text\"))\ntry:\n    mixed.unpivot(\"shop\", [\"tea\", \"cake_text\"], \"item\", \"value\").collect()\nexcept AnalysisException as error:\n    assert error.getCondition() == \"UNPIVOT_VALUE_DATA_TYPE_MISMATCH\"\n    print(error.getCondition())\nelse:\n    raise AssertionError(\"Expected incompatible value types to fail analysis.\")\n\nassert long_units.filter(\"shop = 'North' AND item = 'tea'\").first().units == 3\nassert sales.filter(\"shop = 'North' AND item = 'tea'\").count() == 2",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Qual è la differenza tra il conteggio delle righe e quello delle righe con ricavo?\n2. Come distingui un vero null per il negozio dal totale generale?\n3. Quali dettagli puoi recuperare trasformando i totali delle unità con unpivot?\n\n### Risposte\n\n1. Il primo conta tutte le righe e il secondo solo quelle con ricavo presente.\n2. Controlla gli indicatori di raggruppamento. Il totale generale esclude entrambe le dimensioni.\n3. Recuperi i totali per categoria e negozio. Le quantità delle singole righe e gli identificativi si perdono nell’aggregazione.",
      "id": "cell-17"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}