{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 11 · Calcolare con le funzioni finestra\n\nConfronta osservazioni vicine, classifica le righe e calcola indicatori progressivi mantenendo la grana originale. Usa PySpark classico 4.0.x e Java 17+. In una nuova sessione CPU su Kaggle, se serve, installa PySpark con `%pip install \"pyspark>=4.0,<4.1\"`, poi usa Run All. I dati sono definiti nel notebook.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.sql import SparkSession, Window\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"11-window-functions\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nsales = spark.createDataFrame(\n    [(1, \"North\", 1, 10), (2, \"North\", 2, 20), (3, \"North\", 2, 30),\n     (4, \"North\", 5, 20), (5, \"South\", 1, 7), (6, \"South\", 4, 9)],\n    \"sale_id INT, shop STRING, day_number INT, units INT\",\n)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Funzione finestra\n\nUna funzione finestra calcola un valore per ogni riga usando altre righe della finestra. Un aggregato per gruppo accorpa le righe, mentre un aggregato su finestra le conserva.\n\n## 2. Definire una finestra\n\n`partitionBy(\"shop\")` limita i confronti allo stesso negozio. `orderBy(\"day_number\", \"sale_id\")` stabilisce una sequenza univoca nel negozio. Quest’ordine governa il calcolo. Per controllare la visualizzazione, ordina il DataFrame.\n\n## 3. Intervallo della finestra\n\nL’intervallo è il sottoinsieme della partizione ordinata disponibile per la riga corrente alle funzioni che lo usano. Cambia al cambiare della riga corrente.\n\n## 4. Intervallo per posizione\n\n`rowsBetween(-1, 0)` include la riga precedente e quella corrente quando entrambe esistono.\n\n## 5. Intervallo per valore\n\n`rangeBetween(-1, 0)` sulla chiave numerica del giorno include le righe dal giorno corrente meno uno al giorno corrente. Le due vendite del giorno 2 vedono lo stesso intervallo. Scostamenti limitati come questi richiedono una sola espressione di ordinamento numerica.\n\n## 6. Limiti dell’intervallo\n\nEntrambi gli estremi sono inclusi. Usa `Window.currentRow`, `Window.unboundedPreceding` e `Window.unboundedFollowing` per indicare la posizione corrente e gli estremi senza limite.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "by_shop = Window.partitionBy(\"shop\")\nby_sale = by_shop.orderBy(\"day_number\", \"sale_id\")\nby_day = by_shop.orderBy(\"day_number\")\nprevious_row = by_sale.rowsBetween(-1, Window.currentRow)\nprevious_day = by_day.rangeBetween(-1, Window.currentRow)\n\nframes = sales.select(\n    \"sale_id\", \"shop\", \"day_number\",\n    F.sum(\"units\").over(previous_row).alias(\"last_two_rows\"),\n    F.sum(\"units\").over(previous_day).alias(\"last_two_days\"),\n)\nsecond = frames.filter(\"sale_id = 2\").first()\nassert (second.last_two_rows, second.last_two_days) == (30, 60)\nframes.orderBy(\"sale_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 7. Intervallo predefinito con orderBy\n\nPer un aggregato su finestra ordinata senza intervallo esplicito, Spark usa un intervallo crescente dall’inizio della partizione fino al valore di ordinamento corrente. Le righe a pari merito condividono l’intervallo.\n\n## 8. Ultimo valore nella finestra\n\n`last_value()` legge l’ultimo valore nell’intervallo corrente. Per ottenere l’ultima vendita del negozio, specifica un intervallo completo e un ordinamento univoco. Per la prima vendita di North i due intervalli danno risultati diversi.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "running_rows = by_sale.rowsBetween(Window.unboundedPreceding, Window.currentRow)\nfull_rows = by_sale.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)\nframe_defaults = sales.select(\n    \"sale_id\",\n    F.sum(\"units\").over(by_day).alias(\"default_sum\"),\n    F.last_value(\"units\").over(running_rows).alias(\"last_so_far\"),\n    F.last_value(\"units\").over(full_rows).alias(\"last_in_shop\"),\n)\nfirst = frame_defaults.filter(\"sale_id = 1\").first()\nassert (first.last_so_far, first.last_in_shop) == (10, 20)\nassert [r.default_sum for r in frame_defaults.filter(\"sale_id IN (2, 3)\").collect()] == [60, 60]\nframe_defaults.orderBy(\"sale_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 9. Classifiche\n\n`rank()` assegna lo stesso rango ai pari merito e lascia salti. `dense_rank()` assegna lo stesso rango ai pari merito senza lasciare salti.\n\n## 10. Posizione relativa e fasce\n\n`percent_rank()` riscalibra il rango a partire da zero usando la dimensione della partizione. `cume_dist()` misura la quota di righe fino al valore di ordinamento corrente, inclusi i pari merito. `ntile(2)` divide le righe ordinate in due fasce di dimensione quasi uguale.\n\n## 11. Pari merito con row_number\n\n`row_number()` assegna sempre posizioni distinte. Aggiungi una chiave univoca di spareggio per renderle stabili. Escludila dalla finestra di `rank()` se a quantità uguali deve corrispondere lo stesso rango.\n\n## 12. Le prime N righe per gruppo\n\nCalcola i numeri di riga in una finestra e poi filtrali. Ottieni al massimo N righe per negozio. Un filtro sul rango può conservarne di più in caso di pari merito.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "by_units = by_shop.orderBy(F.col(\"units\").desc())\nby_units_stable = by_shop.orderBy(F.col(\"units\").desc(), \"sale_id\")\nranked = sales.select(\n    \"*\",\n    F.rank().over(by_units).alias(\"rank\"),\n    F.dense_rank().over(by_units).alias(\"dense_rank\"),\n    F.percent_rank().over(by_units).alias(\"percent_rank\"),\n    F.cume_dist().over(by_units).alias(\"cume_dist\"),\n    F.ntile(2).over(by_units_stable).alias(\"bucket\"),\n    F.row_number().over(by_units_stable).alias(\"position\"),\n)\nranked.orderBy(\"shop\", \"position\").show()\ntop_two = ranked.filter(\"position <= 2\")\nassert {r.sale_id for r in top_two.collect()} == {2, 3, 5, 6}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 13. Valore precedente e successivo\n\nUsa `lag()` per una riga precedente e `lead()` per una successiva. Con scostamento uno e ripiego predefinito, la prima riga non ha valore precedente e l’ultima non ha valore successivo.\n\n## 14. lag conta le righe\n\nUno scostamento di uno significa una riga. Per North il giorno 5 segue il giorno 2, quindi la riga precedente può risalire a diversi giorni prima. Anche due righe dello stesso giorno possono essere adiacenti.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "neighbors = sales.select(\n    \"*\",\n    F.lag(\"units\").over(by_sale).alias(\"previous_units\"),\n    F.lead(\"units\").over(by_sale).alias(\"next_units\"),\n    F.lag(\"day_number\").over(by_sale).alias(\"previous_day\"),\n)\nassert neighbors.filter(\"sale_id = 4\").first().previous_day == 2\nneighbors.orderBy(\"sale_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Confronto anno su anno\n\nUna riga precedente rappresenta l’anno precedente solo se lo conferma la chiave temporale. Qui ogni riga rappresenta un negozio in un anno. Prima della differenza, verifica che l’anno osservato precedente sia quello corrente meno uno.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "annual = spark.createDataFrame(\n    [(\"North\", 2022, 80), (\"North\", 2024, 100), (\"North\", 2025, 125)],\n    \"shop STRING, year INT, units INT\",\n)\nyear_order = Window.partitionBy(\"shop\").orderBy(\"year\")\nannual_previous = annual.select(\n    \"*\",\n    F.lag(\"year\").over(year_order).alias(\"previous_year\"),\n    F.lag(\"units\").over(year_order).alias(\"previous_units\"),\n)\nannual_comparison = annual_previous.withColumn(\n    \"yoy_units\",\n    F.when(F.col(\"previous_year\") == F.col(\"year\") - 1, F.col(\"units\") - F.col(\"previous_units\")),\n)\nassert {r.year: r.yoy_units for r in annual_comparison.collect()} == {2022: None, 2024: None, 2025: 25}\nannual_comparison.orderBy(\"year\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 16. Media mobile\n\nUna media mobile sulle ultime tre righe include quella corrente e fino a due righe precedenti dello stesso negozio.\n\n## 17. Somma cumulata\n\nUn intervallo senza limite inferiore fino alla riga corrente dà un totale progressivo. L’ordinamento univoco rende visibile il contributo di ogni vendita anche quando due condividono il giorno.\n\n## 18. Prime righe con meno valori precedenti\n\nLa finestra della prima riga contiene una riga, quella della seconda due. Conta le osservazioni e richiedi un intervallo pieno se l’indicatore richiede tre righe.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "trailing = by_sale.rowsBetween(-2, Window.currentRow)\nprogress = sales.select(\n    \"sale_id\",\n    F.avg(\"units\").over(trailing).alias(\"moving_average\"),\n    F.count(\"*\").over(trailing).alias(\"frame_rows\"),\n    F.sum(\"units\").over(running_rows).alias(\"cumulative_units\"),\n)\nassert [(r.frame_rows, r.cumulative_units) for r in progress.orderBy(\"sale_id\").collect()] == [\n    (1, 10), (2, 30), (3, 60), (3, 80), (1, 7), (2, 16)\n]\nprogress.orderBy(\"sale_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 19. Quota sul gruppo\n\nDividi le unità della vendita per il totale del negozio. Il denominatore usa una finestra senza ordinamento che copre tutto il negozio. `try_divide()` dà null quando il totale è zero.\n\n## 20. orderBy nella finestra della quota\n\nL’ordinamento cambia l’intervallo predefinito. Nell’esempio ordinato il denominatore è cumulato, quindi la prima vendita di North ha quota 1 invece della sua quota sul totale del negozio. Se occorre ordinare, usa un intervallo completo esplicito per il denominatore.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "shares = sales.select(\n    \"sale_id\", \"shop\",\n    F.try_divide(\"units\", F.sum(\"units\").over(by_shop)).alias(\"shop_share\"),\n    F.try_divide(\"units\", F.sum(\"units\").over(by_sale)).alias(\"running_share\"),\n)\nfirst_share = shares.filter(\"sale_id = 1\").first()\nassert first_share.shop_share == 0.125\nassert first_share.running_share == 1.0\nshares.orderBy(\"sale_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 21. Costo di una finestra\n\nSpark spesso deve riunire le chiavi uguali e ordinare le righe in ogni partizione. Cerca `Exchange`, `Sort` e `Window` nel piano. Gruppi grandi o sbilanciati possono dominare il tempo di esecuzione.\n\n## 22. Finestra senza partitionBy\n\nUna classifica globale ha una sola partizione logica. Spark può spostare tutti i dati in una sola partizione fisica. Esamina il piano sui piccoli dati di prova e riserva le finestre globali ai casi che richiedono tutte le righe insieme.",
      "id": "cell-17"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "ranked.explain()\nglobal_rank = sales.withColumn(\"global_position\", F.row_number().over(Window.orderBy(\"sale_id\")))\nglobal_rank.explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-18"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché le due vendite di North del giorno 2 condividono l’intervallo per valore?\n2. Come selezioni esattamente due righe per negozio quando ogni negozio ne ha almeno due?\n3. Perché aggiungere orderBy può cambiare una quota sul gruppo?\n\n### Risposte\n\n1. L’intervallo usa il valore del giorno, quindi giorni uguali condividono i limiti.\n2. Usa `row_number()` con uno spareggio univoco e filtra le posizioni 1 e 2.\n3. La finestra aggregata ordinata usa per impostazione predefinita un intervallo crescente, quindi il denominatore può diventare un totale progressivo.",
      "id": "cell-19"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}