{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 05 · Scegliere le colonne\n\nSeleziona, rinomina e calcola colonne mantenendo le righe di input. La lezione è autonoma e usa PySpark classico 4.x e Java 17 o successivo. In una nuova sessione CPU su Kaggle, se serve, installa PySpark in una cella di preparazione con `%pip install \"pyspark>=4.0,<4.1\"`, poi usa **Run All**. I dati sono definiti qui sotto.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\nfrom pyspark.sql.types import NumericType\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"05-choose-columns\").getOrCreate()\nsales = spark.createDataFrame(\n    [(101, \" tea \", 2, 250, \" counter \"), (102, \"cake\", 1, 375, \" takeaway \")],\n    \"order_id INT, item STRING, units INT, unit_price_cents INT, sales_channel STRING\",\n)\nsales.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Selezionare colonne per nome\n\nPassa i nomi a `select()` nell’ordine desiderato per l’output. L’anteprima conserva una riga per ogni riga d’ordine di partenza.\n\n## 2. Rinominare nella selezione\n\nUsa `alias()` su un’espressione di colonna per assegnarle il nome di output.\n\n## 3. Indicare una colonna\n\n`F.col(\"units\")` crea un riferimento che Spark risolve rispetto all’input. `sales[\"units\"]` fa riferimento a una colonna di questo DataFrame. Puoi passare una stringa a metodi che accettano nomi di colonne, come `select()`.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "sales.select(\"item\", \"units\").show()\nlabels = sales.select(sales[\"item\"].alias(\"product\"), F.col(\"units\").alias(\"quantity\"))\nassert labels.columns == [\"product\", \"quantity\"]\nlabels.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 4. Rinominare tutte le colonne per posizione\n\n`toDF()` assegna una lista completa di nomi alle posizioni delle colonne esistenti. Passa un nome per ogni colonna di input.\n\n## 5. Eliminare colonne\n\n`drop()` conserva le colonne rimanenti. Usalo quando vuoi eliminare alcuni campi noti.\n\n## 6. Eliminare una colonna con un nome inesistente\n\nEliminare una colonna indicando un nome inesistente come stringa lascia lo schema com’era. Verifica che i campi di input attesi esistano se la loro assenza segnalerebbe un problema nei dati.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "new_names = [\"receipt_id\", \"product\", \"quantity\", \"price_cents\", \"channel\"]\nassert sales.toDF(*new_names).columns == new_names\nassert sales.drop(\"sales_channel\").columns == [\"order_id\", \"item\", \"units\", \"unit_price_cents\"]\nassert sales.drop(\"missing_column\").columns == sales.columns",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 7. Rinominare una colonna\n\n`withColumnRenamed()` associa un nome esistente a un nome nuovo mantenendo le altre colonne.\n\n## 8. Rinominare più colonne\n\n`withColumnsRenamed()` accetta una corrispondenza tra nomi esistenti e nuovi nomi. Usa nomi di output distinti per evitare riferimenti ambigui.\n\n## 9. Nome generato di una colonna calcolata\n\nSpark può generare un nome di output da un’espressione. Assegna un alias all’espressione se altro codice dovrà riferirsi al risultato.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "assert sales.withColumnRenamed(\"item\", \"product\").columns[1] == \"product\"\nrenamed = sales.withColumnsRenamed({\"item\": \"product\", \"units\": \"quantity\"})\nassert renamed.columns == [\"order_id\", \"product\", \"quantity\", \"unit_price_cents\", \"sales_channel\"]\n\nline_total = F.col(\"units\") * F.col(\"unit_price_cents\")\nprint(\"Generated name:\", sales.select(line_total).columns)\nsales.select(line_total.alias(\"line_total_cents\")).show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 10. toDF con colonne in ordine diverso\n\nQuando cambia l’ordine di input, la stessa lista di nomi può etichettare i valori sbagliati. Qui entrambi i campi scambiati sono interi, quindi anche un controllo dei tipi mancherebbe l’errore. Rinomina per nome quando l’ordine di input può variare.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "reordered = sales.select(\"units\", \"item\", \"order_id\", \"unit_price_cents\", \"sales_channel\")\nwrong = reordered.toDF(*new_names)\nassert {row.receipt_id for row in wrong.select(\"receipt_id\").collect()} == {1, 2}\nfixed = reordered.withColumnsRenamed(dict(zip(sales.columns, new_names)))\nassert {row.receipt_id for row in fixed.select(\"receipt_id\").collect()} == {101, 102}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 11. Selezionare colonne per tipo\n\nUsa i campi dello schema per trovare le colonne con tipi numerici Spark. Tra queste ci sono anche gli identificativi numerici. Verifica il significato di ogni colonna prima di trattarla come misura.\n\n## 12. Selezionare colonne con una regola sul nome\n\nFiltra la lista dei nomi delle colonne con una regola Python sulle stringhe. Qui il suffisso `_cents` identifica le colonne espresse in centesimi.\n\n## 13. Applicare la stessa operazione a più colonne\n\nCostruisci un’espressione per ciascuna colonna da modificare. Qui `trim()` rimuove gli spazi ai bordi di due campi di testo. Il controllo con `isin()` verifica che le etichette dei canali ripulite siano nella lista dei valori ammessi. Le condizioni sul testo tornano nel Notebook 06.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "numeric_names = [field.name for field in sales.schema if isinstance(field.dataType, NumericType)]\nassert numeric_names == [\"order_id\", \"units\", \"unit_price_cents\"]\nsales.select(*numeric_names).show()\n\ncent_names = [name for name in sales.columns if name.endswith(\"_cents\")]\nassert cent_names == [\"unit_price_cents\"]\n\ntext_names = {\"item\", \"sales_channel\"}\ncleaned = sales.select(*[\n    F.trim(name).alias(name) if name in text_names else F.col(name)\n    for name in sales.columns\n])\ncleaned.show()\nvalid_channel = F.col(\"sales_channel\").isin(\"counter\", \"takeaway\")\nassert cleaned.filter(valid_channel).count() == 2",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 14. Nodo Project nel piano\n\nUsa `explain()` per vedere come Spark rappresenta la selezione. Cerca un nodo `Project` che calcola le espressioni selezionate. Spark può semplificare le proiezioni, quindi il piano stampato dipende dalla query e dalla sua ottimizzazione. Il Notebook 20 tratta i piani in dettaglio.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "cleaned.select(\"order_id\", line_total.alias(\"line_total_cents\")).explain()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Aggiungere più colonne insieme\n\nPassa a `withColumns()` una corrispondenza tra nomi ed espressioni. Entrambe le espressioni qui sotto usano colonne già presenti nell’input. Se una seconda espressione richiede una colonna appena calcolata, rendi esplicita la dipendenza con un passaggio separato.\n\n## 16. withColumn in un ciclo\n\nOgni chiamata a `withColumn()` aggiunge una proiezione al piano. Ripeterla molte volte può ingrandire il piano logico, anche se Spark in seguito combina le proiezioni. Per espressioni indipendenti, usa una sola chiamata a `withColumns()` o `select()`. Il confronto verifica che i due approcci diano gli stessi valori.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "additions = {\n    \"line_total_cents\": line_total,\n    \"large_line\": F.col(\"units\") >= 2,\n}\none_step = cleaned.withColumns(additions)\nloop = cleaned\nfor name, expression in additions.items():\n    loop = loop.withColumn(name, expression)\n\ncomparison = [\"order_id\", \"line_total_cents\", \"large_line\"]\nassert one_step.select(*comparison).orderBy(\"order_id\").collect() == (\n    loop.select(*comparison).orderBy(\"order_id\").collect()\n)\none_step.select(*comparison).orderBy(\"order_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Come puoi tenere solo `item` e rinominarla in `product`?\n2. Perché riutilizzare una lista di nomi per posizione può etichettare male dati riordinati senza produrre un errore?\n3. Come puoi aggiungere due colonne derivate indipendenti in una sola operazione?\n\n### Risposte\n\n1. Usa `sales.select(F.col(\"item\").alias(\"product\"))`.\n2. `toDF()` assegna i nomi per posizione. I campi scambiati possono avere lo stesso tipo.\n3. Passa a `withColumns()` una corrispondenza tra nomi ed espressioni.",
      "id": "cell-17"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}