{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 16 · Scrivere dati\n\nScrivi file e tabelle di catalogo, poi verifica il risultato. Usa una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se serve, installa con `%pip install \"pyspark>=4.0,<4.1\"`. La lezione crea destinazioni temporanee.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pathlib import Path\nfrom tempfile import TemporaryDirectory\nfrom uuid import uuid4\n\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\n\nworkspace = TemporaryDirectory(prefix=\"spark-nb16-\")\nroot = Path(workspace.name)\nspark = (\n    SparkSession.builder.master(\"local[2]\").appName(\"16-write-data\")\n    .config(\"spark.sql.catalogImplementation\", \"in-memory\")\n    .config(\"spark.sql.warehouse.dir\", (root / \"warehouse\").as_uri())\n    .getOrCreate()\n)\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nsales = spark.createDataFrame(\n    [(101, 2, \"North\"), (102, 3, \"South\"), (103, 1, \"North\"), (104, 4, \"South\")],\n    \"order_id INT, units INT, shop STRING\",\n)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Catalogo\n\nIl catalogo permette a Spark di risolvere i nomi delle tabelle e consultare metadati come schema e posizione. Puoi leggere un percorso direttamente, senza una voce di catalogo. Qui si usa un catalogo in memoria locale alla sessione.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "database = \"nb16_\" + uuid4().hex[:12]\nspark.sql(f\"CREATE DATABASE {database} LOCATION '{(root / database).as_uri()}'\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 2. Scrivere in Parquet\n\nUsa `df.write.parquet(path)` per scrivere un dataset Parquet. La destinazione è di solito una directory con file di dati.\n\n## 3. Ridurre le partizioni senza rimescolare\n\n`coalesce(1)` riduce le partizioni Spark senza aggiungere uno shuffle. Qui i dati sono pochi. Un unico task di scrittura può rallentare output grandi.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "parquet = str(root / \"sales_parquet\")\nsales.coalesce(1).write.parquet(parquet)\nassert spark.read.parquet(parquet).count() == 4",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 4. Modalità di salvataggio\n\nLa modalità predefinita `errorifexists` genera un errore se la destinazione esiste. `ignore` la salta, `append` aggiunge dati e `overwrite` la sostituisce secondo le regole della sorgente.\n\n## 5. Forma generale della scrittura\n\nUsa `write.format(...).mode(...).options(...).save(path)` per specificare formato e opzioni.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "batch_path = str(root / \"batches\")\nsales.limit(1).write.format(\"parquet\").mode(\"errorifexists\").save(batch_path)\nsales.write.format(\"parquet\").mode(\"ignore\").save(batch_path)\nassert spark.read.parquet(batch_path).count() == 1\nsales.write.format(\"parquet\").mode(\"append\").save(batch_path)\nassert spark.read.parquet(batch_path).count() == 5",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 6. Formati per righe e per colonne\n\nIl CSV codifica record completi come righe di testo. Parquet memorizza i valori per colonna all’interno di gruppi di righe. I lettori compatibili possono così saltare le colonne inutili.\n\n## 7. Il CSV non conserva i tipi\n\nIl CSV non incorpora lo schema Spark. In lettura, fornisci uno schema o deducilo. In assenza di entrambi, l’ID numerico viene letto come stringa.\n\n## 8. File e partizioni in memoria\n\nIn una scrittura semplice senza partizionamento su disco, ogni task con dati scrive di norma un file. Partizioni vuote, limiti di record e partizionamento in directory possono cambiare il conteggio. Ignora i file marcatori nell’ispezione.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "csv_path = str(root / \"sales_csv\")\nsales.coalesce(1).write.option(\"header\", True).csv(csv_path)\nassert spark.read.option(\"header\", True).csv(csv_path).schema[\"order_id\"].dataType.simpleString() == \"string\"\nassert len(list(Path(parquet).glob(\"part-*.parquet\"))) == 1",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 9. Partizionare i file in directory\n\n`partitionBy(\"shop\")` colloca le righe in directory come `shop=North`. Queste partizioni su disco sono distinte dalle partizioni Spark in memoria.\n\n## 10. Overwrite può sostituire tutta la destinazione\n\nQuesta sovrascrittura Parquet senza partizionamento sostituisce l’intero dataset temporaneo. Le quattro righe originali lasciano il posto alla sola riga selezionata. Verifica il percorso prima di sovrascrivere.\n\n## 11. Righe massime per file\n\n`maxRecordsPerFile` può suddividere l’output di un task in più file. Limita il numero di righe, indipendentemente dai byte.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "partitioned = str(root / \"by_shop\")\nsales.coalesce(1).write.partitionBy(\"shop\").parquet(partitioned)\nassert (Path(partitioned) / \"shop=North\").is_dir()\n\nsales.filter(\"order_id = 101\").write.mode(\"overwrite\").parquet(parquet)\nassert [r.order_id for r in spark.read.parquet(parquet).collect()] == [101]\n\nlimited = root / \"limited\"\nsales.coalesce(1).write.option(\"maxRecordsPerFile\", 2).parquet(str(limited))\nassert len(list(limited.glob(\"part-*.parquet\"))) == 2",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 12. Potatura delle partizioni\n\nUn filtro sulla chiave di partizione su disco può permettere a Spark di saltare altre directory. Cerca il filtro di partizione nel piano di lettura.\n\n## 13. File piccoli\n\nMolti file piccoli aumentano il costo di elencarli, aprirli e gestire i task. Controlla conteggi e dimensioni quando scegli il numero di task di scrittura.\n\n## 14. Ripartire per colonna\n\n`repartition(2, \"shop\")` rimescola le righe per collocare le chiavi di negozio uguali nella stessa partizione Spark. Combinalo con il partizionamento in directory quando serve quella disposizione.\n\n## 15. Regole per la colonna di partizione\n\nScegli una colonna usata spesso nei filtri, con cardinalità gestibile e dati sufficienti per valore. Un ID ordine univoco creerebbe una directory per ordine. Considera valori mancanti e chiavi sbilanciate.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "north = spark.read.parquet(partitioned).filter(F.col(\"shop\") == \"North\")\nassert north.count() == 2\nnorth.explain()\nprint([(file.name, file.stat().st_size) for file in sorted(limited.glob(\"part-*.parquet\"))])\nsales.repartition(2, \"shop\").write.partitionBy(\"shop\").parquet(str(root / \"clustered\"))",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 16. Salvare una tabella nel catalogo\n\n`saveAsTable()` scrive i dati e registra una tabella. Senza percorso esplicito, qui usa la posizione gestita del database.\n\n## 17. Elencare le tabelle del catalogo\n\n`spark.catalog.listTables(database)` elenca le tabelle visibili nel database.\n\n## 18. Leggere una tabella per nome\n\nUsa `spark.table()` con il nome qualificato. Spark consulta il catalogo per trovare i dati.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "managed = f\"{database}.sales\"\nsales.write.format(\"parquet\").saveAsTable(managed)\nassert \"sales\" in {table.name for table in spark.catalog.listTables(database)}\nassert spark.table(managed).count() == 4",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 19. Inserire righe in una tabella\n\n`insertInto()` scrive in una tabella esistente. Qui accoda una riga.\n\n## 20. insertInto abbina per posizione\n\nLo schema di destinazione è già fissato. Spark assegna i campi in ingresso per posizione. Due interi scambiati possono superare i controlli di tipo e finire nei campi sbagliati. L’esempio temporaneo mostra l’errore, poi allinea le colonne prima del secondo inserimento.",
      "id": "cell-17"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "extra = spark.createDataFrame([(105, 5, \"North\")], sales.schema)\nswapped = extra.select(\"units\", \"order_id\", \"shop\")\nswapped.write.insertInto(managed)\nbad = spark.table(managed).filter(\"order_id = 5\").first()\nassert bad.units == 105\nswapped.select(*spark.table(managed).columns).write.insertInto(managed)\nassert spark.table(managed).filter(\"order_id = 105\").first().units == 5",
      "execution_count": null,
      "outputs": [],
      "id": "cell-18"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 21. Tabella con percorso esplicito\n\nRegistra la tabella indicando il percorso dei dati. In questo esempio i file rimangono dopo aver eliminato la tabella esterna.\n\n## 22. Creare una tabella con SQL\n\n`CREATE TABLE ... USING parquet` dichiara una tabella di catalogo. Qui una tabella gestita di prova riceve una riga via SQL.\n\n## 23. Svuotare una tabella\n\n`TRUNCATE TABLE` rimuove le righe da una tabella gestita compatibile e ne conserva la definizione. L’esempio verifica entrambe le proprietà prima di rimuovere il database temporaneo.",
      "id": "cell-19"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "external = f\"{database}.external_sales\"\nexternal_path = root / \"external_sales\"\nsales.write.format(\"parquet\").option(\"path\", external_path.as_uri()).saveAsTable(external)\nspark.sql(f\"DROP TABLE {external}\")\nassert spark.read.parquet(str(external_path)).count() == 4\n\nsql_table = f\"{database}.sql_sales\"\nspark.sql(f\"CREATE TABLE {sql_table} (order_id INT, units INT) USING parquet\")\nspark.sql(f\"INSERT INTO {sql_table} VALUES (101, 2)\")\nassert spark.table(sql_table).count() == 1\nspark.sql(f\"TRUNCATE TABLE {sql_table}\")\nassert spark.table(sql_table).count() == 0\nassert spark.table(sql_table).columns == [\"order_id\", \"units\"]\nspark.sql(f\"DROP DATABASE {database} CASCADE\")\nworkspace.cleanup()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-20"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Quale operazione crea le directory di partizione?\n2. Perché insertInto può accettare colonne intere scambiate?\n3. Che cosa conserva TRUNCATE?\n\n### Risposte\n\n1. `DataFrameWriter.partitionBy()` crea directory dai valori delle chiavi.\n2. Abbina i campi per posizione e i valori sono compatibili con i tipi di destinazione.\n3. La definizione della tabella nel catalogo, incluso lo schema.",
      "id": "cell-21"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}