11 · Calcolare con le funzioni finestra

Confronta 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.

from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[2]").appName("11-window-functions").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")
sales = spark.createDataFrame(
    [(1, "North", 1, 10), (2, "North", 2, 20), (3, "North", 2, 30),
     (4, "North", 5, 20), (5, "South", 1, 7), (6, "South", 4, 9)],
    "sale_id INT, shop STRING, day_number INT, units INT",
)

1. Funzione finestra

Una 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.

2. Definire una finestra

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.

3. Intervallo della finestra

L’intervallo è il sottoinsieme della partizione ordinata disponibile per la riga corrente alle funzioni che lo usano. Cambia al cambiare della riga corrente.

4. Intervallo per posizione

rowsBetween(-1, 0) include la riga precedente e quella corrente quando entrambe esistono.

5. Intervallo per valore

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.

6. Limiti dell’intervallo

Entrambi gli estremi sono inclusi. Usa Window.currentRow, Window.unboundedPreceding e Window.unboundedFollowing per indicare la posizione corrente e gli estremi senza limite.

by_shop = Window.partitionBy("shop")
by_sale = by_shop.orderBy("day_number", "sale_id")
by_day = by_shop.orderBy("day_number")
previous_row = by_sale.rowsBetween(-1, Window.currentRow)
previous_day = by_day.rangeBetween(-1, Window.currentRow)

frames = sales.select(
    "sale_id", "shop", "day_number",
    F.sum("units").over(previous_row).alias("last_two_rows"),
    F.sum("units").over(previous_day).alias("last_two_days"),
)
second = frames.filter("sale_id = 2").first()
assert (second.last_two_rows, second.last_two_days) == (30, 60)
frames.orderBy("sale_id").show()

7. Intervallo predefinito con orderBy

Per 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.

8. Ultimo valore nella finestra

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.

running_rows = by_sale.rowsBetween(Window.unboundedPreceding, Window.currentRow)
full_rows = by_sale.rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
frame_defaults = sales.select(
    "sale_id",
    F.sum("units").over(by_day).alias("default_sum"),
    F.last_value("units").over(running_rows).alias("last_so_far"),
    F.last_value("units").over(full_rows).alias("last_in_shop"),
)
first = frame_defaults.filter("sale_id = 1").first()
assert (first.last_so_far, first.last_in_shop) == (10, 20)
assert [r.default_sum for r in frame_defaults.filter("sale_id IN (2, 3)").collect()] == [60, 60]
frame_defaults.orderBy("sale_id").show()

9. Classifiche

rank() assegna lo stesso rango ai pari merito e lascia salti. dense_rank() assegna lo stesso rango ai pari merito senza lasciare salti.

10. Posizione relativa e fasce

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.

11. Pari merito con row_number

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.

12. Le prime N righe per gruppo

Calcola 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.

by_units = by_shop.orderBy(F.col("units").desc())
by_units_stable = by_shop.orderBy(F.col("units").desc(), "sale_id")
ranked = sales.select(
    "*",
    F.rank().over(by_units).alias("rank"),
    F.dense_rank().over(by_units).alias("dense_rank"),
    F.percent_rank().over(by_units).alias("percent_rank"),
    F.cume_dist().over(by_units).alias("cume_dist"),
    F.ntile(2).over(by_units_stable).alias("bucket"),
    F.row_number().over(by_units_stable).alias("position"),
)
ranked.orderBy("shop", "position").show()
top_two = ranked.filter("position <= 2")
assert {r.sale_id for r in top_two.collect()} == {2, 3, 5, 6}

13. Valore precedente e successivo

Usa 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.

14. lag conta le righe

Uno 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.

neighbors = sales.select(
    "*",
    F.lag("units").over(by_sale).alias("previous_units"),
    F.lead("units").over(by_sale).alias("next_units"),
    F.lag("day_number").over(by_sale).alias("previous_day"),
)
assert neighbors.filter("sale_id = 4").first().previous_day == 2
neighbors.orderBy("sale_id").show()

15. Confronto anno su anno

Una 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.

annual = spark.createDataFrame(
    [("North", 2022, 80), ("North", 2024, 100), ("North", 2025, 125)],
    "shop STRING, year INT, units INT",
)
year_order = Window.partitionBy("shop").orderBy("year")
annual_previous = annual.select(
    "*",
    F.lag("year").over(year_order).alias("previous_year"),
    F.lag("units").over(year_order).alias("previous_units"),
)
annual_comparison = annual_previous.withColumn(
    "yoy_units",
    F.when(F.col("previous_year") == F.col("year") - 1, F.col("units") - F.col("previous_units")),
)
assert {r.year: r.yoy_units for r in annual_comparison.collect()} == {2022: None, 2024: None, 2025: 25}
annual_comparison.orderBy("year").show()

16. Media mobile

Una media mobile sulle ultime tre righe include quella corrente e fino a due righe precedenti dello stesso negozio.

17. Somma cumulata

Un 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.

18. Prime righe con meno valori precedenti

La 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.

trailing = by_sale.rowsBetween(-2, Window.currentRow)
progress = sales.select(
    "sale_id",
    F.avg("units").over(trailing).alias("moving_average"),
    F.count("*").over(trailing).alias("frame_rows"),
    F.sum("units").over(running_rows).alias("cumulative_units"),
)
assert [(r.frame_rows, r.cumulative_units) for r in progress.orderBy("sale_id").collect()] == [
    (1, 10), (2, 30), (3, 60), (3, 80), (1, 7), (2, 16)
]
progress.orderBy("sale_id").show()

19. Quota sul gruppo

Dividi 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.

20. orderBy nella finestra della quota

L’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.

shares = sales.select(
    "sale_id", "shop",
    F.try_divide("units", F.sum("units").over(by_shop)).alias("shop_share"),
    F.try_divide("units", F.sum("units").over(by_sale)).alias("running_share"),
)
first_share = shares.filter("sale_id = 1").first()
assert first_share.shop_share == 0.125
assert first_share.running_share == 1.0
shares.orderBy("sale_id").show()

21. Costo di una finestra

Spark 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.

22. Finestra senza partitionBy

Una 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.

ranked.explain()
global_rank = sales.withColumn("global_position", F.row_number().over(Window.orderBy("sale_id")))
global_rank.explain()

Verifica la comprensione

1. Perché le due vendite di North del giorno 2 condividono l’intervallo per valore?
2. Come selezioni esattamente due righe per negozio quando ogni negozio ne ha almeno due?
3. Perché aggiungere orderBy può cambiare una quota sul gruppo?

Risposte

1. L’intervallo usa il valore del giorno, quindi giorni uguali condividono i limiti.
2. Usa row_number() con uno spareggio univoco e filtra le posizioni 1 e 2.
3. La finestra aggregata ordinata usa per impostazione predefinita un intervallo crescente, quindi il denominatore può diventare un totale progressivo.