10 · Aggregare e cambiare forma

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

import pandas as pd
from pyspark.errors import AnalysisException
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[2]").appName("10-aggregate-and-reshape").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")
sales = spark.createDataFrame(
    [(1, "North", "tea", 2, 500),
     (2, "North", "tea", 1, 250),
     (3, "North", "cake", 1, None),
     (4, "South", "tea", 3, 750),
     (5, None, "cake", 2, 700)],
    "line_id INT, shop STRING, item STRING, units INT, revenue_cents INT",
)
sales.orderBy("line_id").show()

1. Raggruppare le righe

groupBy("shop") definisce i gruppi per negozio. Anche le righe con negozio null formano un gruppo.

2. Calcolare aggregati per gruppo

agg() calcola una riga di risultato per gruppo. Qui la somma conta le unità alla grana del negozio.

3. Contare i valori distinti

count_distinct("item") conta i nomi di prodotto distinti e non null per negozio.

4. Funzioni di aggregazione

Funzioni come sum(), avg(), min() e max() riassumono più righe in un valore per gruppo. Un alias assegna un nome a ogni risultato.

5. Contare righe e valori presenti

count("*") conta tutte le righe. count("revenue_cents") conta quelle con un ricavo presente.

6. Conteggio distinto approssimato

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.

7. count(colonna) ignora i null

North ha tre righe e solo due valori di ricavo registrati. Usa entrambi i conteggi per mostrare la copertura.

by_shop = sales.groupBy("shop").agg(
    F.sum("units").alias("units_sold"),
    F.count_distinct("item").alias("exact_items"),
    F.avg("units").alias("mean_units"),
    F.count("*").alias("lines"),
    F.count("revenue_cents").alias("priced_lines"),
    F.approx_count_distinct("item").alias("estimated_items"),
)
by_shop.orderBy(F.col("shop").asc_nulls_last()).show()
north = by_shop.filter("shop = 'North'").first()
assert (north.units_sold, north.exact_items, north.lines, north.priced_lines) == (4, 2, 3, 2)

8. Mediana e percentili

Usa median() per la mediana e percentile_approx() per quantili approssimati. Un percentile approssimato può scegliere un valore osservato laddove la mediana esatta interpolerebbe.

9. Raccogliere i valori in una lista

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.

10. Aggregazione globale

sales.agg() aggrega l’intero DataFrame e restituisce una riga.

11. Raggruppare per un’espressione

Una chiave di gruppo può essere un’espressione. Qui units >= 2 divide le righe in gruppi da una unità e da più unità.

12. Aggregazione condizionale

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

sales.agg(
    F.median("units").alias("median_units"),
    F.percentile_approx("units", [0.25, 0.75]).alias("quartiles"),
    F.sort_array(F.collect_list("item")).alias("items"),
).show(truncate=False)
sales.groupBy((F.col("units") >= 2).alias("multi_unit")).count().show()
tea_units = F.sum(F.when(F.col("item") == "tea", F.col("units")).otherwise(0))
assert sales.agg(tea_units).first()[0] == 6
sales.groupBy("shop").agg(tea_units.alias("tea_units")).show()

13. Subtotali

rollup("shop", "item") aggiunge ai gruppi negozio-prodotto i subtotali per negozio e il totale generale.

14. Null nelle righe di subtotale

Spark rappresenta con null una dimensione esclusa dal raggruppamento del subtotale. Anche una riga di dettaglio può avere un vero null per il negozio.

15. Riconoscere le righe di subtotale

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.

16. Tutte le combinazioni di subtotali

cube() include ogni sottoinsieme delle dimensioni di raggruppamento. Con negozio e prodotto aggiunge i subtotali per solo prodotto alle combinazioni di rollup.

def summarize(grouped):
    return grouped.agg(
        F.sum("units").alias("units_sold"),
        F.grouping("shop").alias("all_shops"),
        F.grouping("item").alias("all_items"),
    )

rolled = summarize(sales.rollup("shop", "item"))
cubed = summarize(sales.cube("shop", "item"))
rolled.orderBy("all_shops", "all_items", "shop", "item").show()
assert rolled.filter("all_shops = 1 AND all_items = 1").first().units_sold == 9
assert rolled.filter("shop IS NULL AND all_shops = 0 AND all_items = 0").count() == 1
assert cubed.filter("all_shops = 1 AND all_items = 0").count() == 2

17. Aggregazione personalizzata

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

18. Gruppo intero nella memoria dell’executor

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

19. Calcolare un indicatore per gruppo

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

@F.pandas_udf("double")
def multi_share(values: pd.Series) -> float:
    return float((values >= 2).mean())

custom = sales.groupBy("shop").agg(multi_share("units").alias("share"))
native = sales.groupBy("shop").agg(F.avg((F.col("units") >= 2).cast("int")).alias("share"))
assert {r.shop: r.share for r in custom.collect()} == {r.shop: r.share for r in native.collect()}
native.show()

20. Pivot

pivot() sposta le categorie di prodotto nelle colonne. Raggruppa prima per negozio e scegli la misura da aggregare in ogni cella.

21. Pivot senza elenco di valori

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

22. count("*") dentro un pivot

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.

23. Celle vuote dopo il pivot

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

24. Più misure nel pivot

Passa più aggregati con alias ad agg() per ottenere una colonna per categoria e misura.

items = ["tea", "cake"]
wide = sales.groupBy("shop").pivot("item", items).sum("units")
wide.orderBy("shop").show()
discovered = sales.groupBy("shop").pivot("item").sum("units")
assert set(discovered.columns) == {"shop", "tea", "cake"}

metrics = sales.groupBy("shop").pivot("item", items).agg(
    F.count("*").alias("lines"),
    F.count("revenue_cents").alias("priced"),
    F.sum("revenue_cents").alias("revenue"),
)
north_metrics = metrics.filter("shop = 'North'").first()
assert (north_metrics.cake_lines, north_metrics.cake_priced, north_metrics.cake_revenue) == (1, 0, None)
assert wide.filter("shop = 'South'").first().cake is None
metrics.show()

25. fillna su più colonne

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

wide_units = wide.fillna(0, subset=items)
assert wide_units.filter("shop = 'South'").first().cake == 0
wide_units.show()

26. Unpivot

unpivot() trasforma colonne di misure in coppie etichetta-valore. Mantieni le colonne identificative e assegna un nome alle nuove colonne per categoria e valore.

27. Unpivot con stack

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.

28. Tipi diversi nell’unpivot

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

29. Pivot e unpivot non sono sempre inversi

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

long_units = wide_units.unpivot("shop", items, "item", "units")
stacked = wide_units.selectExpr("shop", "stack(2, 'tea', tea, 'cake', cake) AS (item, units)")
assert {(r.shop, r.item, r.units) for r in long_units.collect()} == {
    (r.shop, r.item, r.units) for r in stacked.collect()
}
long_units.orderBy("shop", "item").show()

mixed = wide_units.select("shop", "tea", F.col("cake").cast("string").alias("cake_text"))
try:
    mixed.unpivot("shop", ["tea", "cake_text"], "item", "value").collect()
except AnalysisException as error:
    assert error.getCondition() == "UNPIVOT_VALUE_DATA_TYPE_MISMATCH"
    print(error.getCondition())
else:
    raise AssertionError("Expected incompatible value types to fail analysis.")

assert long_units.filter("shop = 'North' AND item = 'tea'").first().units == 3
assert sales.filter("shop = 'North' AND item = 'tea'").count() == 2

Verifica la comprensione

1. Qual è la differenza tra il conteggio delle righe e quello delle righe con ricavo?
2. Come distingui un vero null per il negozio dal totale generale?
3. Quali dettagli puoi recuperare trasformando i totali delle unità con unpivot?

Risposte

1. Il primo conta tutte le righe e il secondo solo quelle con ricavo presente.
2. Controlla gli indicatori di raggruppamento. Il totale generale esclude entrambe le dimensioni.
3. Recuperi i totali per categoria e negozio. Le quantità delle singole righe e gli identificativi si perdono nell’aggregazione.