{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 09 · Usare tipi complessi\n\nIspeziona e trasforma dati d’ordine annidati con array, mappe e struct. Infine interpreta JSON come struct tipizzata o VARIANT. Esegui tutte le celle in una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa PySpark in una cella di preparazione con `%pip install \"pyspark>=4.0,<4.1\"`. Tutti i dati sono definiti nel notebook.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"09-use-complex-types\").getOrCreate()\nspark.conf.set(\"spark.sql.ansi.enabled\", \"true\")\norders = spark.createDataFrame(\n    [\n        (101, [\"tea\", \"cake\", \"tea\"], {\"tea\": 2, \"cake\": 1}, (\"counter\", 4)),\n        (102, [], {}, (\"takeaway\", 0)),\n        (103, None, None, None),\n    ],\n    \"order_id INT, items ARRAY<STRING>, quantities MAP<STRING,INT>, pickup STRUCT<channel:STRING, wait_minutes:INT>\",\n)\norders.printSchema()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Array\n\nUn array è una raccolta ordinata di elementi dello stesso tipo. Può contenere duplicati, come nell’ordine 101.\n\n## 2. Separare un array in righe\n\n`explode()` produce una riga per elemento. Gli array null e vuoti non producono righe.\n\n## 3. Dimensione di un array\n\n`array_size()` restituisce il numero di elementi, zero per un array vuoto e null per un array null.\n\n## 4. Leggere un elemento di un array\n\n`get()` legge l’elemento con indice a partire da zero e restituisce null per un indice fuori intervallo.\n\n## 5. Cercare un valore\n\n`array_contains()` verifica l’appartenenza. Un array null dà null.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "items = F.col(\"items\")\nexploded = orders.select(\"order_id\", F.explode(items).alias(\"item\"))\nassert exploded.count() == 3\nexploded.show()\norders.select(\n    \"order_id\", F.array_size(items).alias(\"size\"),\n    F.get(items, 0).alias(\"first_item\"), F.array_contains(items, \"tea\").alias(\"has_tea\"),\n).orderBy(\"order_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 6. Operazioni su insiemi tra array\n\n`array_union()`, `array_intersect()` e `array_except()` restituiscono gli elementi distinti dell’unione, intersezione e differenza. Tratta i risultati come insiemi quando l’ordine è irrilevante.\n\n## 7. element_at conta da 1 e [i] da 0\n\nPer gli array, `element_at(items, 1)` e `items[0]` leggono entrambi il primo elemento. Un indice negativo in `element_at` conta dalla fine. In modalità ANSI una posizione non valida può generare un errore. `try_element_at()` restituisce null per posizioni fuori intervallo, mentre l’indice zero resta non valido.\n\n## 8. Unire gli elementi in un testo\n\n`array_join()` costruisce una stringa con un separatore. Salta gli elementi null a meno che tu non fornisca un sostituto.\n\n## 9. Funzioni di ordine superiore\n\n`transform()` applica un’espressione a ogni elemento, `filter()` seleziona gli elementi e `aggregate()` li combina in un valore. Le lambda Python costruiscono espressioni di colonna Spark.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "one_order = orders.filter(\"order_id = 101\")\nmenu = F.array(F.lit(\"tea\"), F.lit(\"coffee\"))\none_order.select(\n    F.array_union(items, menu).alias(\"union\"),\n    F.array_intersect(items, menu).alias(\"intersection\"),\n    F.array_except(items, menu).alias(\"difference\"),\n).show()\nindexes = one_order.select(F.element_at(items, 1), items[0], F.element_at(items, -1)).first()\nassert tuple(indexes) == (\"tea\", \"tea\", \"tea\")\n\none_order.select(F.array_join(items, \", \").alias(\"label\")).show()\norders.select(\n    \"order_id\",\n    F.transform(items, lambda item: F.upper(item)).alias(\"upper_items\"),\n    F.filter(items, lambda item: item == \"tea\").alias(\"teas\"),\n    F.aggregate(items, F.lit(0), lambda count, item: count + F.length(item)).alias(\"characters\"),\n).show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 10. Mappa\n\nUna mappa memorizza valori indicizzati da chiavi univoche e non null. La mappa delle quantità indica il numero di unità ordinate per prodotto.\n\n## 11. Creare una mappa\n\n`create_map()` accetta espressioni alternate di chiave e valore. Racchiudi le chiavi e i valori letterali in `lit()`.\n\n## 12. Chiavi e valori di una mappa\n\n`map_keys()` e `map_values()` restituiscono array senza ordine garantito. Usa `map_entries()` per mantenere ogni chiave associata al suo valore.\n\n## 13. Leggere il valore di una chiave\n\n`element_at(map, key)` restituisce il valore corrispondente o null se la chiave manca.\n\n## 14. Separare una mappa in righe\n\n`explode()` emette una chiave e un valore per ogni voce.\n\n## 15. Valori di una mappa dello stesso tipo\n\nUna mappa Spark ha un tipo dichiarato per le chiavi e uno per i valori. Usa una struct se servono campi con nome e tipi diversi.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "quantities = F.col(\"quantities\")\ndefaults = F.create_map(F.lit(\"tea\"), F.lit(0), F.lit(\"cake\"), F.lit(0))\norders.select(\"order_id\", defaults.alias(\"defaults\"), F.map_keys(quantities), F.map_values(quantities)).show()\nassert tuple(one_order.select(F.element_at(quantities, \"tea\"), F.element_at(quantities, \"coffee\")).first()) == (2, None)\nentries = orders.select(\"order_id\", F.explode(quantities).alias(\"item\", \"units\"))\nassert {(r.item, r.units) for r in entries.collect()} == {(\"tea\", 2), (\"cake\", 1)}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 16. Struttura (struct)\n\nUna struct raggruppa campi con nome, ognuno con il proprio tipo. Qui il ritiro contiene un canale testuale e un tempo di attesa intero. `struct()` costruisce una struttura da espressioni.\n\n## 17. Leggere un campo di una struttura\n\nUsa percorsi con punti in `select()` oppure `getField()` su una colonna struct.\n\n## 18. Portare i campi in colonne separate\n\n`pickup.*` espande i campi immediati della struttura. Il risultato mantiene una riga per ordine di input.\n\n## 19. Aggiungere o togliere campi di una struttura\n\n`withField()` aggiunge o sostituisce un campo. `dropFields()` rimuove i campi indicati. Una struct genitore null rimane null.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "orders.select(\"order_id\", F.struct(\"order_id\", \"pickup\").alias(\"order_details\")).show(truncate=False)\nassert one_order.select(F.col(\"pickup\").getField(\"channel\")).first()[0] == \"counter\"\nexpanded = orders.select(\"order_id\", \"pickup.*\")\nassert expanded.columns == [\"order_id\", \"channel\", \"wait_minutes\"]\nedited = orders.select(\n    \"order_id\",\n    F.col(\"pickup\").withField(\"confirmed\", F.lit(True)).dropFields(\"wait_minutes\").alias(\"pickup\"),\n)\nedited.printSchema()\nedited.show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 20. Tenere array vuoti e null\n\n`explode_outer()` emette una riga segnaposto con elemento null per ogni array vuoto o null. Conserva la dimensione originale per distinguere i due casi.\n\n## 21. Posizione dell’elemento\n\n`posexplode_outer()` aggiunge una posizione a partire da zero. Le righe segnaposto hanno posizione null.\n\n## 22. explode cambia la grana\n\nL’input ha una riga per ordine. L’output espanso ha una riga per elemento, più le righe segnaposto con la variante outer. Contare le righe espanse dà un risultato diverso dal numero di ordini.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "outer = orders.select(\"order_id\", F.array_size(items).alias(\"original_size\"),\n                      F.explode_outer(items).alias(\"item\"))\nassert outer.count() == 5\npositioned = orders.select(\"order_id\", F.posexplode_outer(items).alias(\"position\", \"item\"))\npositioned.orderBy(\"order_id\", \"position\").show()\nassert [r.position for r in positioned.filter(\"order_id = 101\").orderBy(\"position\").collect()] == [0, 1, 2]",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 23. Leggere testo JSON in una colonna\n\n`from_json()` interpreta il testo secondo uno schema. Il risultato qui sotto è una struct che Spark può ispezionare campo per campo.\n\n## 24. Scrivere una struttura come JSON\n\n`to_json()` serializza la struttura in una stringa. Serializza i campi interpretati, che possono differire dal contenuto originale.\n\n## 25. Estrarre un campo da testo JSON\n\n`get_json_object()` legge un percorso dal testo originale. La colonna restituita è una stringa.\n\n## 26. VARIANT\n\nSpark 4 aggiunge VARIANT per dati semistrutturati con forma variabile. `try_parse_json()` restituisce null SQL per JSON malformato e `try_variant_get()` legge un percorso con il tipo richiesto, restituendo null se il percorso manca o la conversione fallisce.\n\n## 27. from_json con schema incompatibile\n\nIn modalità permissiva, un oggetto JSON valido può dare campi null se i nomi o i tipi differiscono da quelli dichiarati. Conserva l’input originale e verifica i campi obbligatori. Un campo null da solo non distingue uno schema incompatibile da un campo assente o da un null JSON esplicito.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "payloads = spark.createDataFrame(\n    [(1, '{\"channel\":\"counter\",\"wait_minutes\":4}'),\n     (2, '{\"channel\":\"takeaway\",\"wait_minutes\":\"unknown\"}'),\n     (3, 'not JSON')],\n    \"payload_id INT, raw_json STRING\",\n)\nschema = \"channel STRING, wait_minutes INT\"\nparsed = payloads.withColumn(\"pickup\", F.from_json(\"raw_json\", schema, {\"mode\": \"PERMISSIVE\"}))\nparsed.select(\"payload_id\", \"pickup.*\", F.to_json(\"pickup\").alias(\"serialized\"),\n              F.get_json_object(\"raw_json\", \"$.wait_minutes\").alias(\"wait_text\")).show(truncate=False)\n\nvariants = payloads.withColumn(\"payload\", F.try_parse_json(\"raw_json\"))\nwait = variants.select(\"payload_id\", F.try_variant_get(\"payload\", \"$.wait_minutes\", \"int\").alias(\"wait\"))\nassert {r.payload_id: r.wait for r in wait.collect()} == {1: 4, 2: None, 3: None}\n\nwaits = parsed.select(\"payload_id\", F.col(\"pickup.wait_minutes\").alias(\"wait\")).collect()\nassert {r.payload_id: r.wait for r in waits} == {1: 4, 2: None, 3: None}\nwrong_schema = payloads.filter(\"payload_id = 1\").select(\n    F.from_json(\"raw_json\", \"wait_seconds INT\").getField(\"wait_seconds\").alias(\"wait_seconds\")\n)\nassert wrong_schema.first().wait_seconds is None",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Come conservi un ordine con array vuoto quando espandi i prodotti?\n2. Quale tipo annidato permette al canale e al tempo di attesa di avere tipi diversi?\n3. Perché conservare il JSON originale quando verifichi una conversione tipizzata?\n\n### Risposte\n\n1. Usa `explode_outer()` e conserva la dimensione originale se devi distinguere array vuoti e null.\n2. Usa una struct con campi distinti e dotati di nome.\n3. Un null tipizzato può avere più cause. Il contenuto originale consente di indagarle.",
      "id": "cell-15"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}