{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 07 · Calcolare nuove colonne\n\nRicava 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.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "import pandas as pd\nfrom pyspark.errors import PySparkException\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"07-compute-new-columns\").getOrCreate()\nspark.conf.set(\"spark.sql.ansi.enabled\", \"true\")\norders = spark.createDataFrame(\n    [(101, \" Tea \", \"SKU-014\", 2, 250, 1),\n     (102, \"COFFEE\", \"SKU-207\", 5, 375, 0),\n     (103, None, \"unknown\", 1, 400, None)],\n    \"order_id INT, item STRING, sku STRING, units INT, price_cents INT, guests INT\",\n)\nunits = F.col(\"units\")\nprice = F.col(\"price_cents\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Aggiungere o sostituire una colonna\n\n`withColumn()` restituisce un nuovo DataFrame. Un nome nuovo aggiunge una colonna. Un nome esistente sostituisce la sua espressione nel risultato.\n\n## 2. Valore costante in una colonna\n\n`lit()` crea un’espressione con lo stesso valore su ogni riga.\n\n## 3. Generare numeri casuali\n\n`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.\n\n## 4. Arrotondare\n\n`round(x, 2)` arrotonda a due decimali con HALF_UP. Usa un letterale decimale per rappresentare esattamente il caso a metà.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from decimal import Decimal\n\nlabeled = orders.withColumn(\"source\", F.lit(\"cafe\"))\nrevised = labeled.withColumn(\"source\", F.lit(\"cafe_demo\"))\nassert \"source\" not in orders.columns\nrevised.select(\"order_id\", \"source\", F.rand(7).alias(\"draw\")).show()\nassert orders.select(F.round(F.lit(Decimal(\"2.345\")), 2)).first()[0] == Decimal(\"2.35\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 5. Operazioni aritmetiche tra colonne\n\nMoltiplica il numero di unità per il prezzo unitario in centesimi. Usa un operando intero lungo per ampliare l’intervallo del calcolo.\n\n## 6. Funzioni matematiche\n\nApplica funzioni integrate come `sqrt()` e `abs()` direttamente alle colonne.\n\n## 7. Limitare un valore tra due estremi\n\nCombina `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.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "total = units.cast(\"bigint\") * price\nguest = F.col(\"guests\")\nclamped = F.when(guest.isNotNull(), F.least(F.greatest(guest, F.lit(1)), F.lit(3)))\norders.select(\"order_id\", total.alias(\"total_cents\"), F.sqrt(units).alias(\"sqrt_units\"),\n              F.abs(units - 3).alias(\"distance_from_three\"), clamped.alias(\"clamped_guests\")).show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 8. Divisione per zero\n\nCon ANSI abilitata, una divisione ordinaria per zero genera un errore quando Spark la valuta. Intercetta l’errore atteso per proseguire.\n\n## 9. Divisione sicura\n\n`try_divide()` restituisce null se il divisore è zero. Anche un divisore null produce null. Conserva il divisore originale per distinguere i due casi.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "try:\n    orders.select((total / guest).alias(\"cents_per_guest\")).collect()\nexcept PySparkException as error:\n    assert error.getCondition() == \"DIVIDE_BY_ZERO\"\n    print(error.getCondition())\nelse:\n    raise AssertionError(\"Expected division by zero to fail.\")\n\nsafe = orders.select(\"order_id\", \"guests\", F.try_divide(total, guest).alias(\"cents_per_guest\"))\nassert {r.order_id: r.cents_per_guest for r in safe.collect()} == {101: 500.0, 102: None, 103: None}\nsafe.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 10. Cambiare maiuscole e togliere spazi\n\nCombina `trim()` e `lower()` per normalizzare le etichette di esempio.\n\n## 11. Estrarre una sottostringa\n\nUsa `substring()` per estrarre una parte dello SKU di lunghezza fissa.\n\n## 12. Espressioni regolari\n\n`regexp_extract()` estrae il gruppo di cattura richiesto dalla prima corrispondenza. Con un input non null senza corrispondenze, Spark restituisce una stringa vuota.\n\n## 13. Unire testi\n\n`concat()` combina espressioni di tipo stringa. Racchiudi un separatore fisso in `lit()`.\n\n## 14. substring conta da 1\n\nLa posizione 1 è il primo carattere. La posizione 5 è la prima cifra in `SKU-014`.\n\n## 15. concat con un null\n\nPer 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.\n\n## 16. Lunghezza e riempimento del testo\n\n`length()` conta i caratteri di una stringa. `lpad()` aggiunge caratteri a sinistra fino alla lunghezza indicata e tronca le stringhe più lunghe.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "item = F.lower(F.trim(\"item\"))\ntext_columns = orders.select(\n    \"order_id\",\n    item.alias(\"clean_item\"),\n    F.substring(\"sku\", 1, 3).alias(\"sku_prefix\"),\n    F.regexp_extract(\"sku\", r\"^SKU-([0-9]+)$\", 1).alias(\"sku_digits\"),\n    F.concat(item, F.lit(\" / \"), F.col(\"sku\")).alias(\"strict_label\"),\n    F.concat_ws(\" / \", item, F.col(\"sku\")).alias(\"display_label\"),\n    F.substring(\"sku\", 5, 3).alias(\"fixed_slice\"),\n    F.length(item).alias(\"item_length\"),\n    F.lpad(F.col(\"order_id\").cast(\"string\"), 5, \"0\").alias(\"receipt_label\"),\n)\ntext_columns.orderBy(\"order_id\").show(truncate=False)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 17. Logica condizionale\n\nUsa `when()` per assegnare valori in base a condizioni sulla riga e `otherwise()` per il caso predefinito.\n\n## 18. Ordine dei rami di when\n\nSpark sceglie il primo ramo corrispondente. Se le soglie si sovrappongono, controlla prima quella più specifica.\n\n## 19. when senza otherwise\n\nSe manca un caso predefinito, le righe che non soddisfano alcuna condizione ricevono null.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "size = F.when(units >= 5, \"bulk\").when(units >= 2, \"multi\").otherwise(\"single\")\nflag = F.when(units >= 5, \"review\")\nclassified = orders.select(\"order_id\", size.alias(\"size\"), flag.alias(\"flag\"))\nassert {r.order_id: (r.size, r.flag) for r in classified.collect()} == {\n    101: (\"multi\", None), 102: (\"bulk\", \"review\"), 103: (\"single\", None)\n}\nclassified.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 20. Data e ora correnti\n\n`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.\n\n## 21. Identificativo univoco\n\n`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.\n\n## 22. Identificativo non consecutivo\n\nSpark codifica informazioni sulla partizione negli ID generati, quindi possono esserci salti. Un nuovo partizionamento può cambiare l’ID assegnato a una riga.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "orders.select(F.current_date().alias(\"run_date\"), F.current_timestamp().alias(\"run_time\")).show(1)\nids = spark.range(6, numPartitions=2).select(\"id\", F.monotonically_increasing_id().alias(\"generated_id\"))\nvalues = [r.generated_id for r in ids.orderBy(\"id\").collect()]\nassert len(set(values)) == 6\nprint(values)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 23. Funzione personalizzata (UDF)\n\nUna 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.\n\n## 24. Nodo ArrowEvalPython nel piano\n\nArrow è 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.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def normalize(value):\n    return value.strip().lower() if value is not None else None\n\nplain = F.udf(normalize, \"string\", useArrow=False)\narrow = F.udf(normalize, \"string\", useArrow=True)\norders.select(arrow(\"item\").alias(\"normalized\")).explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 25. pandas UDF\n\nUna pandas UDF scalare riceve blocchi come Series pandas e restituisce una Series della stessa lunghezza. Spark scambia i blocchi tramite Arrow.\n\n## 26. Logica della UDF opaca all’ottimizzatore\n\nCatalyst 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.",
      "id": "cell-17"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "@F.pandas_udf(\"string\")\ndef normalize_batch(values: pd.Series) -> pd.Series:\n    return values.str.strip().str.lower()\n\ncompared = orders.select(\n    item.alias(\"native\"),\n    plain(\"item\").alias(\"plain\"),\n    arrow(\"item\").alias(\"arrow\"),\n    normalize_batch(\"item\").alias(\"pandas\"),\n)\nfor row in compared.collect():\n    assert row.native == row.plain == row.arrow == row.pandas\norders.select(item.alias(\"native\")).explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-18"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 27. Arrow per impostazione predefinita nelle UDF Python\n\nSe 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.",
      "id": "cell-19"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "print(\"Scalar Python UDF Arrow setting:\", spark.conf.get(\"spark.sql.execution.pythonUDF.arrow.enabled\"))\ndefault_normalize = F.udf(normalize, \"string\")\norders.select(default_normalize(\"item\").alias(\"normalized\")).explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-20"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 28. Funzione che restituisce una tabella (UDTF)\n\nUna 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.",
      "id": "cell-21"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "@F.udtf(returnType=\"position INT, token STRING\", useArrow=False)\nclass Tokens:\n    def eval(self, label: str):\n        for position, token in enumerate(label.split(), start=1):\n            yield position, token\n\ntokens = Tokens(F.lit(\"iced tea\"))\nassert [(r.position, r.token) for r in tokens.orderBy(\"position\").collect()] == [(1, \"iced\"), (2, \"tea\")]\ntokens.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-22"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché conservare il denominatore dopo una divisione sicura?\n2. Che cosa succede se entrambe le soglie sono soddisfatte?\n3. Quale esempio può restituire più righe con una sola chiamata?\n\n### Risposte\n\n1. Permette di distinguere un denominatore zero da uno mancante.\n2. Il primo ramo corrispondente fornisce il valore.\n3. La funzione che restituisce una tabella emette una riga per parola.",
      "id": "cell-23"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}