07 · Calcolare nuove colonne

Ricava misure, etichette pulite e categorie dalle righe d’ordine del bar. La lezione 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 con Internet abilitato. L’extra SQL include pandas e PyArrow per gli esempi vettoriali. Poi usa Run All.

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

spark = SparkSession.builder.master("local[2]").appName("07-compute-new-columns").getOrCreate()
spark.conf.set("spark.sql.ansi.enabled", "true")
orders = spark.createDataFrame(
    [(101, " Tea ", "SKU-014", 2, 250, 1),
     (102, "COFFEE", "SKU-207", 5, 375, 0),
     (103, None, "unknown", 1, 400, None)],
    "order_id INT, item STRING, sku STRING, units INT, price_cents INT, guests INT",
)
units = F.col("units")
price = F.col("price_cents")

1. Aggiungere o sostituire una colonna

withColumn() restituisce un nuovo DataFrame. Un nome nuovo aggiunge una colonna. Un nome esistente sostituisce la sua espressione nel risultato.

2. Valore costante in una colonna

lit() crea un’espressione con lo stesso valore su ogni riga.

3. Generare numeri casuali

rand(seed) genera double pseudocasuali nell’intervallo [0, 1). Il seme è utile con una disposizione di esecuzione fissa. La stessa chiave di business può ricevere un valore diverso se cambia tale disposizione.

4. Arrotondare

round(x, 2) arrotonda a due decimali con HALF_UP. Usa un letterale decimale per rappresentare esattamente il caso a metà.

from decimal import Decimal

labeled = orders.withColumn("source", F.lit("cafe"))
revised = labeled.withColumn("source", F.lit("cafe_demo"))
assert "source" not in orders.columns
revised.select("order_id", "source", F.rand(7).alias("draw")).show()
assert orders.select(F.round(F.lit(Decimal("2.345")), 2)).first()[0] == Decimal("2.35")

5. Operazioni aritmetiche tra colonne

Moltiplica il numero di unità per il prezzo unitario in centesimi. Usa un operando intero lungo per ampliare l’intervallo del calcolo.

6. Funzioni matematiche

Applica funzioni integrate come sqrt() e abs() direttamente alle colonne.

7. Limitare un valore tra due estremi

Combina greatest() e least() per limitare il valore a [1, 3]. Spark ignora gli argomenti null in queste funzioni. Usa una condizione esplicita se vuoi mantenere i valori mancanti.

total = units.cast("bigint") * price
guest = F.col("guests")
clamped = F.when(guest.isNotNull(), F.least(F.greatest(guest, F.lit(1)), F.lit(3)))
orders.select("order_id", total.alias("total_cents"), F.sqrt(units).alias("sqrt_units"),
              F.abs(units - 3).alias("distance_from_three"), clamped.alias("clamped_guests")).show()

8. Divisione per zero

Con ANSI abilitata, una divisione ordinaria per zero genera un errore quando Spark la valuta. Intercetta l’errore atteso per proseguire.

9. Divisione sicura

try_divide() restituisce null se il divisore è zero. Anche un divisore null produce null. Conserva il divisore originale per distinguere i due casi.

try:
    orders.select((total / guest).alias("cents_per_guest")).collect()
except PySparkException as error:
    assert error.getCondition() == "DIVIDE_BY_ZERO"
    print(error.getCondition())
else:
    raise AssertionError("Expected division by zero to fail.")

safe = orders.select("order_id", "guests", F.try_divide(total, guest).alias("cents_per_guest"))
assert {r.order_id: r.cents_per_guest for r in safe.collect()} == {101: 500.0, 102: None, 103: None}
safe.show()

10. Cambiare maiuscole e togliere spazi

Combina trim() e lower() per normalizzare le etichette di esempio.

11. Estrarre una sottostringa

Usa substring() per estrarre una parte dello SKU di lunghezza fissa.

12. Espressioni regolari

regexp_extract() estrae il gruppo di cattura richiesto dalla prima corrispondenza. Con un input non null senza corrispondenze, Spark restituisce una stringa vuota.

13. Unire testi

concat() combina espressioni di tipo stringa. Racchiudi un separatore fisso in lit().

14. substring conta da 1

La posizione 1 è il primo carattere. La posizione 5 è la prima cifra in SKU-014.

15. concat con un null

Per queste stringhe, concat() restituisce null se un input è null. concat_ws() salta le stringhe null, il che può essere utile per comporre etichette da visualizzare.

16. Lunghezza e riempimento del testo

length() conta i caratteri di una stringa. lpad() aggiunge caratteri a sinistra fino alla lunghezza indicata e tronca le stringhe più lunghe.

item = F.lower(F.trim("item"))
text_columns = orders.select(
    "order_id",
    item.alias("clean_item"),
    F.substring("sku", 1, 3).alias("sku_prefix"),
    F.regexp_extract("sku", r"^SKU-([0-9]+)$", 1).alias("sku_digits"),
    F.concat(item, F.lit(" / "), F.col("sku")).alias("strict_label"),
    F.concat_ws(" / ", item, F.col("sku")).alias("display_label"),
    F.substring("sku", 5, 3).alias("fixed_slice"),
    F.length(item).alias("item_length"),
    F.lpad(F.col("order_id").cast("string"), 5, "0").alias("receipt_label"),
)
text_columns.orderBy("order_id").show(truncate=False)

17. Logica condizionale

Usa when() per assegnare valori in base a condizioni sulla riga e otherwise() per il caso predefinito.

18. Ordine dei rami di when

Spark sceglie il primo ramo corrispondente. Se le soglie si sovrappongono, controlla prima quella più specifica.

19. when senza otherwise

Se manca un caso predefinito, le righe che non soddisfano alcuna condizione ricevono null.

size = F.when(units >= 5, "bulk").when(units >= 2, "multi").otherwise("single")
flag = F.when(units >= 5, "review")
classified = orders.select("order_id", size.alias("size"), flag.alias("flag"))
assert {r.order_id: (r.size, r.flag) for r in classified.collect()} == {
    101: ("multi", None), 102: ("bulk", "review"), 103: ("single", None)
}
classified.show()

20. Data e ora correnti

current_date() e current_timestamp() usano l’inizio della valutazione della query. Le chiamate alla stessa funzione in una query coincidono. I valori mostrati dipendono dal momento di esecuzione e dal fuso della sessione.

21. Identificativo univoco

monotonically_increasing_id() può assegnare interi lunghi univoci in una valutazione, entro i limiti documentati da Spark per righe e partizioni. Mantieni la chiave di business per l’identità nel tempo.

22. Identificativo non consecutivo

Spark codifica informazioni sulla partizione negli ID generati, quindi possono esserci salti. Un nuovo partizionamento può cambiare l’ID assegnato a una riga.

orders.select(F.current_date().alias("run_date"), F.current_timestamp().alias("run_time")).show(1)
ids = spark.range(6, numPartitions=2).select("id", F.monotonically_increasing_id().alias("generated_id"))
values = [r.generated_id for r in ids.orderBy("id").collect()]
assert len(set(values)) == 6
print(values)

23. Funzione personalizzata (UDF)

Una UDF scalare Python incapsula una funzione con un tipo Spark dichiarato per il risultato. La logica è volutamente semplice per poterla confrontare con un’espressione nativa.

24. Nodo ArrowEvalPython nel piano

Arrow è una rappresentazione colonnare usata per trasferire dati tra Spark e Python. In Spark classico 4.x cerca ArrowEvalPython nel piano fisico dell’esempio con Arrow. L’opzione esplicita rende l’esempio indipendente dall’impostazione predefinita.

def normalize(value):
    return value.strip().lower() if value is not None else None

plain = F.udf(normalize, "string", useArrow=False)
arrow = F.udf(normalize, "string", useArrow=True)
orders.select(arrow("item").alias("normalized")).explain()

25. pandas UDF

Una pandas UDF scalare riceve blocchi come Series pandas e restituisce una Series della stessa lunghezza. Spark scambia i blocchi tramite Arrow.

26. Logica della UDF opaca all’ottimizzatore

Catalyst può pianificare dove valutare una UDF Python, ma non può ispezionarne il corpo per riscriverlo con espressioni native. Qui basta lower(trim(...)). Il suo comportamento e la rimozione degli spazi in Python possono differire per altri caratteri di spaziatura.

@F.pandas_udf("string")
def normalize_batch(values: pd.Series) -> pd.Series:
    return values.str.strip().str.lower()

compared = orders.select(
    item.alias("native"),
    plain("item").alias("plain"),
    arrow("item").alias("arrow"),
    normalize_batch("item").alias("pandas"),
)
for row in compared.collect():
    assert row.native == row.plain == row.arrow == row.pandas
orders.select(item.alias("native")).explain()

27. Arrow per impostazione predefinita nelle UDF Python

Se ometti useArrow, l’impostazione di sessione spark.sql.execution.pythonUDF.arrow.enabled controlla le UDF scalari ordinarie. Il valore predefinito è false in Spark 4.0 e diventa true in Spark 4.2. Leggi il valore effettivo della sessione. Le opzioni esplicite sopra rendono stabile il confronto.

print("Scalar Python UDF Arrow setting:", spark.conf.get("spark.sql.execution.pythonUDF.arrow.enabled"))
default_normalize = F.udf(normalize, "string")
orders.select(default_normalize("item").alias("normalized")).explain()

28. Funzione che restituisce una tabella (UDTF)

Una UDTF Python può emettere zero o più righe per chiamata. Dichiara lo schema di output e produci tuple da eval(). Qui viene emessa una riga per parola dell’etichetta. La chiamata diretta usa un valore letterale.

@F.udtf(returnType="position INT, token STRING", useArrow=False)
class Tokens:
    def eval(self, label: str):
        for position, token in enumerate(label.split(), start=1):
            yield position, token

tokens = Tokens(F.lit("iced tea"))
assert [(r.position, r.token) for r in tokens.orderBy("position").collect()] == [(1, "iced"), (2, "tea")]
tokens.show()

Verifica la comprensione

1. Perché conservare il denominatore dopo una divisione sicura?
2. Che cosa succede se entrambe le soglie sono soddisfatte?
3. Quale esempio può restituire più righe con una sola chiamata?

Risposte

1. Permette di distinguere un denominatore zero da uno mancante.
2. Il primo ramo corrispondente fornisce il valore.
3. La funzione che restituisce una tabella emette una riga per parola.