{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 02 · Creare una tabella\n\nCrea un DataFrame Spark da dati locali, scegli lo schema e converti testo irregolare in valori tipizzati. La lezione è autonoma e usa PySpark classico 4.0.x, Python 3.9+, Java 17+ e pandas 2+. Su Kaggle, se serve, esegui `%pip install \"pyspark>=4.0,<4.1\" \"pandas>=2\" scikit-learn` in una cella di preparazione con Internet abilitato. Poi usa **Run All**. I dati Iris sono inclusi in scikit-learn.",
      "id": "cell-01"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Importare i moduli di PySpark\n\nImporta esplicitamente i nomi usati nella lezione. `col()` fa riferimento a una colonna per nome. `lit()` fornisce un valore costante. Entrambe compaiono nell’ultimo esempio.\n\n## 2. SparkSession\n\nUna SparkSession è il punto di accesso per creare DataFrame e impostare le opzioni SQL.\n\n## 3. Import con asterisco\n\nUn import come `from pyspark.sql.functions import *` introduce molti nomi nello spazio dei nomi corrente. Può nascondere nomi Python come `sum`. Gli import espliciti ne rendono visibile la provenienza.\n\n## 4. Creare o riusare una sessione\n\n`getOrCreate()` riusa una sessione disponibile o ne avvia una. Qui, se si parte da zero, `local[2]` richiede un’applicazione Spark locale con due thread di lavoro.",
      "id": "cell-02"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from datetime import date, datetime\nfrom decimal import Decimal\n\nimport pandas as pd\nfrom pyspark.errors import PySparkValueError, PySparkException\nfrom pyspark.sql import Row, SparkSession\nfrom pyspark.sql.functions import col, lit\nfrom pyspark.sql.types import DecimalType, IntegerType, StructField, StructType\nfrom sklearn.datasets import load_iris\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"02-create-a-table\").getOrCreate()\nassert spark.version.startswith(\"4.\"), \"This lesson requires PySpark 4.x.\"",
      "execution_count": null,
      "outputs": [],
      "id": "cell-03"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 5. Caricare una tabella pubblica\n\nCarica Iris in pandas e passa la piccola tabella a Spark. `columns` elenca i nomi. `select()` mantiene le colonne richieste, `alias()` assegna un nome a una colonna di output e `distinct()` conserva una copia di ogni riga distinta. `count()` restituisce il numero di righe.",
      "id": "cell-04"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "iris_pd = load_iris(as_frame=True).frame\niris = spark.createDataFrame(iris_pd)\nprint(iris.columns)\nlabels = iris.select(iris[\"target\"].alias(\"species_id\")).distinct()\nassert labels.count() == 3\nlabels.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-05"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 6. Tipo di dato di una colonna\n\nUna colonna ha un tipo Spark che determina quali valori può contenere e quali operazioni si possono applicare. `printSchema()` mostra i nomi dei campi, i tipi e la possibilità di avere valori mancanti.",
      "id": "cell-06"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "iris.printSchema()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-07"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 7. Riga come oggetto Row\n\n`Row` permette di assegnare un nome ai valori di un record. Qui ogni record contiene una riga d’ordine.\n\n## 8. Creare una tabella da dati Python\n\nPassa a `createDataFrame()` una lista di oggetti Row. Spark può dedurre i tipi delle colonne dai valori Python.\n\n## 9. Creare una sequenza di numeri\n\n`spark.range(3)` restituisce un DataFrame con gli interi lunghi 0, 1 e 2 in una colonna chiamata `id`.",
      "id": "cell-08"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "orders = [Row(order_id=101, drink_name=\"tea\"), Row(order_id=102, drink_name=\"coffee\")]\norders_df = spark.createDataFrame(orders)\norders_df.show()\nspark.range(3).show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-09"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 10. Schema\n\nLo schema descrive i campi, il loro ordine, i tipi e la possibilità di contenere null.\n\n## 11. Nomi delle colonne in snake case\n\nUsa nomi minuscoli con trattini bassi, come `order_id`. `toDF()` sostituisce tutti i nomi delle colonne per posizione. Mantieni i nomi nello stesso ordine delle colonne di partenza.",
      "id": "cell-10"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "orders_df.printSchema()\nrenamed_orders = orders_df.toDF(\"receipt_id\", \"drink_name\")\nassert renamed_orders.columns == [\"receipt_id\", \"drink_name\"]",
      "execution_count": null,
      "outputs": [],
      "id": "cell-11"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 12. Avviare un programma PySpark\n\nIn un notebook le celle vengono eseguite nello stesso kernel Python. In uno script autonomo, inserisci la creazione della sessione e le operazioni sotto `if __name__ == \"__main__\":` e arresta la sessione al termine. Il driver coordina l’applicazione Spark. Qui il kernel del notebook esegue la parte Python del driver.\n\n## 13. Scrivere lo schema come testo\n\nUna stringa DDL specifica per Spark nomi e tipi delle colonne. In questo esempio fissa anche l’ordine dei campi.\n\n## 14. None di Python diventa null in Spark\n\nUn valore mancante fornito come `None` diventa `NULL` in SQL se il campo ammette null.",
      "id": "cell-12"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "quantity_rows = [(101, 2), (102, None)]\nquantities = spark.createDataFrame(quantity_rows, \"order_id INT, quantity INT\")\nquantities.show()\nquantities.printSchema()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-13"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Creare una tabella da pandas\n\nPassa un DataFrame pandas a `createDataFrame()`. Con Iris hai già visto questo passaggio su dati reali. Qui usiamo una piccola tabella pandas con uno schema esplicito.\n\n## 16. Convertire una tabella in pandas\n\n`toPandas()` copia il risultato Spark in un DataFrame pandas nel processo del driver.\n\n## 17. toPandas su dati grandi\n\nL’intero risultato deve entrare nella memoria del driver. Usa il metodo per risultati piccoli, come questa tabella di due righe.\n\n## 18. Il null di Spark diventa NaN in pandas\n\nIn questa conversione, una colonna Spark di interi che ammette null diventa una colonna pandas in virgola mobile con `NaN` al posto del valore mancante. La rappresentazione dei valori mancanti dipende dal tipo risultante in pandas.",
      "id": "cell-14"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "menu_pd = pd.DataFrame({\"drink_name\": [\"tea\", \"coffee\"], \"unit_price_text\": [\"2.50\", \"3.75\"]})\nmenu = spark.createDataFrame(menu_pd, \"drink_name STRING, unit_price_text STRING\")\nmenu.show()\n\nquantity_pd = quantities.toPandas()\nprint(quantity_pd)\nassert quantity_pd[\"quantity\"].isna().sum() == 1",
      "execution_count": null,
      "outputs": [],
      "id": "cell-15"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 19. Creare una tabella da un RDD\n\nUn RDD è una collezione distribuita. In PySpark classico, `spark.sparkContext` dà accesso allo SparkContext. Qui `parallelize()` trasforma una piccola lista Python in un RDD e la stessa stringa DDL ne descrive le righe.",
      "id": "cell-16"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "quantity_rdd = spark.sparkContext.parallelize(quantity_rows)\nfrom_rdd = spark.createDataFrame(quantity_rdd, \"order_id INT, quantity INT\")\nassert from_rdd.count() == 2",
      "execution_count": null,
      "outputs": [],
      "id": "cell-17"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 20. Possibilità di null in una colonna\n\n`StructField` descrive un campo. `nullable=True` consente valori null. Qui `order_id` è obbligatorio e `quantity` può essere sconosciuta.\n\n## 21. nullable=False non corregge i dati\n\nCon questa lista Python e la verifica dello schema attiva, PySpark rifiuta `None` in un campo che non ammette null. Se la regola richiede un ID d’ordine, gestisci il valore mancante prima di creare il DataFrame.",
      "id": "cell-18"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "quantity_schema = StructType([\n    StructField(\"order_id\", IntegerType(), nullable=False),\n    StructField(\"quantity\", IntegerType(), nullable=True),\n])\nspark.createDataFrame(quantity_rows, quantity_schema).printSchema()\n\ntry:\n    spark.createDataFrame([(None, 2)], quantity_schema)\nexcept PySparkValueError as error:\n    assert error.getCondition() == \"FIELD_NOT_NULLABLE_WITH_NAME\"\n    print(error.getCondition())\nelse:\n    raise AssertionError(\"A missing order ID should fail schema verification.\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-19"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 22. Tipi di dato di Spark\n\nTra i tipi comuni ci sono `STRING`, `BOOLEAN`, `INT`, `BIGINT`, `DOUBLE`, `DECIMAL`, `DATE` e `TIMESTAMP`. Una data rappresenta un giorno di calendario. Un timestamp contiene data e ora. L’esempio con il timestamp usa un fuso orario di sessione fisso.\n\n## 23. Schema esplicito\n\nFornisci a Spark i nomi e i tipi dei campi quando fanno parte del contratto dei dati.\n\n## 24. Schema dedotto\n\nSe passi soltanto i nomi delle colonne, Spark deduce i tipi. Lo stesso intero Python diventa in genere `BIGINT` con la deduzione e `INT` se lo dichiari come tale.",
      "id": "cell-20"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "spark.conf.set(\"spark.sql.session.timeZone\", \"UTC\")\nsample = [(101, \"tea\", True, 2.5, date(2026, 11, 4), datetime(2026, 11, 4, 9))]\nnames = [\"order_id\", \"drink_name\", \"is_hot\", \"rating\", \"order_date\", \"ordered_at\"]\ntyped = spark.createDataFrame(\n    sample,\n    \"order_id INT, drink_name STRING, is_hot BOOLEAN, rating DOUBLE, order_date DATE, ordered_at TIMESTAMP_NTZ\",\n)\ninferred = spark.createDataFrame(sample, names)\ntyped.printSchema()\ninferred.printSchema()\nassert typed.schema[\"order_id\"].dataType.simpleString() == \"int\"\nassert inferred.schema[\"order_id\"].dataType.simpleString() == \"bigint\"",
      "execution_count": null,
      "outputs": [],
      "id": "cell-21"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 25. Convertire il tipo di una colonna\n\nUsa `cast()` per convertire una colonna esistente. Il risultato qui sotto è di tipo intero, anche se i valori originali sono stringhe.\n\n## 26. Modalità ANSI\n\nL’impostazione di sessione `spark.sql.ansi.enabled` controlla come Spark gestisce alcune operazioni non valide. Impostala esplicitamente per gli esempi che seguono.",
      "id": "cell-22"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "raw_quantities = spark.createDataFrame([(\"2\",), (\"oops\",), (None,)], \"quantity_text STRING\")\nvalid_quantity = spark.createDataFrame([(\"2\",)], \"quantity_text STRING\")\nvalid_quantity.select(col(\"quantity_text\").cast(\"int\").alias(\"quantity\")).show()\nspark.conf.set(\"spark.sql.ansi.enabled\", \"true\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-23"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 27. Decimal per il denaro\n\nPer gli importi monetari usa tipi decimali a virgola fissa. `DECIMAL(8,2)` consente otto cifre totali, di cui due dopo la virgola. Crea il Decimal Python da una stringa per preservare il valore desiderato.",
      "id": "cell-24"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "money_schema = StructType([StructField(\"unit_price\", DecimalType(8, 2), nullable=False)])\nspark.createDataFrame([(Decimal(\"2.50\"),)], money_schema).show()\nmenu.select(\"drink_name\", col(\"unit_price_text\").cast(\"decimal(8,2)\").alias(\"unit_price\")).show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-25"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 28. Conversione non valida in modalità ANSI\n\nIl testo `oops` non si può convertire in un intero. L’errore emerge quando `show()` valuta la conversione. L’esempio intercetta l’errore previsto e permette a **Run All** di proseguire.",
      "id": "cell-26"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "try:\n    raw_quantities.select(col(\"quantity_text\").cast(\"int\").alias(\"quantity\")).show()\nexcept PySparkException as error:\n    assert error.getCondition() == \"CAST_INVALID_INPUT\"\n    print(error.getCondition())\nelse:\n    raise AssertionError(\"The malformed quantity should fail in ANSI mode.\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-27"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 29. Conversione sicura\n\nIn questa conversione supportata, `try_cast()` restituisce null per il testo malformato. Conserva il valore originale per distinguere il testo malformato da un input mancante. Qui `isNull()` verifica se il risultato manca e `isNotNull()` verifica se l’input è presente. `lit()` aggiunge un’etichetta costante per la provenienza.",
      "id": "cell-28"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "parsed = raw_quantities.select(\n    \"quantity_text\", col(\"quantity_text\").try_cast(\"int\").alias(\"quantity\")\n)\nchecked = parsed.select(\n    \"quantity_text\",\n    \"quantity\",\n    (col(\"quantity_text\").isNotNull() & col(\"quantity\").isNull()).alias(\"invalid_quantity\"),\n    lit(\"cafe_demo\").alias(\"source\"),\n)\nchecked.show()\nassert {row.quantity_text: (row.quantity, row.invalid_quantity) for row in checked.collect()} == {\n    \"2\": (2, False),\n    \"oops\": (None, True),\n    None: (None, False),\n}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-29"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Quale schema assegna il tipo INT a `order_id`?\n2. Che cosa conviene conservare quando una conversione sicura restituisce null per un input malformato?\n3. Perché `toPandas()` è adatto all’esempio di due righe?\n\n### Risposte\n\n1. Lo schema esplicito che dichiara `order_id INT` o usa `IntegerType()`.\n2. Conserva il testo originale o un indicatore della conversione fallita.\n3. Entrambe le righe entrano facilmente nella memoria del driver.",
      "id": "cell-30"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Riferimenti\n\nDocumentazione API di Apache Spark per createDataFrame e Column.try_cast e guida alla conformità ANSI di Spark SQL.",
      "id": "cell-31"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}