{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 15 · Leggere dati\n\nLeggi piccoli file, controlla gli schemi ed esamina i record scartati. Esegui in una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se necessario, installa PySpark con `%pip install \"pyspark[sql]>=4.0,<4.1\"`. Il notebook crea i propri file. L’esempio JDBC richiede un database esterno e si esegue solo se lo abiliti esplicitamente.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "import os\nfrom pathlib import Path\nfrom tempfile import TemporaryDirectory\n\nfrom pyspark.errors import PySparkException\nfrom py4j.protocol import Py4JJavaError\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\nfrom pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"15-read-data\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\nworkspace = TemporaryDirectory(prefix=\"spark-nb15-\")\nroot = Path(workspace.name)\n\ndef fixture(name, content):\n    path = root / name\n    path.parent.mkdir(parents=True, exist_ok=True)\n    path.write_text(content, encoding=\"utf-8\")\n    return str(path)\n\ncsv_path = fixture(\"orders.csv\", \"id,day,units\\n1,2026-11-01,2\\n2,2026-11-02,3\\n\")\njson_path = fixture(\"orders.json\", '{\"id\":1,\"units\":2}\\n{\"id\":2,\"units\":3}\\n')",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Leggere un file CSV in una tabella\n\n`spark.read.csv(path)` legge record delimitati in un DataFrame.\n\n## 2. Usare la prima riga per i nomi\n\nImposta `header=True` quando la prima riga contiene i nomi delle colonne. Altrimenti Spark assegna nomi come `_c0` e tratta la prima riga come dato.\n\n## 3. Dedurre i tipi in lettura\n\nPer impostazione predefinita, i campi CSV sono stringhe. `inferSchema=True` chiede a Spark di esaminare i valori e dedurne i tipi, il che può richiedere un passaggio in più.\n\n## 4. Leggere JSON\n\n`spark.read.json()` legge per impostazione predefinita JSON delimitato da nuove righe. Qui ogni riga è un oggetto JSON distinto.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "assert spark.read.csv(csv_path).count() == 3\nstrings = spark.read.csv(csv_path, header=True)\ninferred = spark.read.csv(csv_path, header=True, inferSchema=True)\nassert strings.schema[\"units\"].dataType.simpleString() == \"string\"\nassert inferred.schema[\"units\"].dataType.simpleString() == \"int\"\nspark.read.json(json_path).orderBy(\"id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 5. Sorgente di dati\n\nUna sorgente indica a Spark come trasformare l’input esterno in righe. CSV, JSON, Parquet e JDBC sono alcuni esempi.\n\n## 6. Forma generale della lettura\n\nUsa `spark.read.format(...).options(...).load(...)` per specificare la sorgente e le sue opzioni.\n\n## 7. Estensione e separatore del file\n\nUn separatore divide i campi. Nel prossimo file è un punto e virgola. L’estensione è una convenzione sul nome del file.\n\n## 8. Imporre lo schema in lettura\n\nPassa uno schema quando conosci i tipi dei campi. Scegli opzioni compatibili con il contratto della sorgente e verifica l’intestazione e i dati letti.\n\n## 9. Quando la deduzione lascia le date come testo\n\nSpark può dedurre DATE da input adatti. Una colonna che sembra contenere date può restare STRING se la deduzione incontra testo incompatibile o se la preferenza per le date è disattivata. Controlla lo schema prima di calcolare con le date.\n\n## 10. Formato delle date in lettura\n\nQuando un campo dichiarato DATE usa un formato personalizzato, imposta `dateFormat` di conseguenza. Qui `dd/MM/yyyy` indica giorno, mese e anno di calendario.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "semicolon = fixture(\"delivery.csv\", \"id;day;units\\n1;01/11/2026;2\\n\")\ntyped = (\n    spark.read.format(\"csv\").options(header=True, sep=\";\", dateFormat=\"dd/MM/yyyy\")\n    .schema(\"id INT, day DATE, units INT\").load(semicolon)\n)\nassert str(typed.first().day) == \"2026-11-01\"\nas_text = spark.read.options(header=True, inferSchema=True, preferDate=False).csv(csv_path)\nassert as_text.schema[\"day\"].dataType.simpleString() == \"string\"",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 11. Leggere un file come testo\n\n`spark.read.text()` restituisce una colonna stringa chiamata `value`, con una riga per linea di testo per impostazione predefinita.\n\n## 12. Opzioni di forma del CSV\n\nOpzioni come `sep`, `quote` e `multiLine` descrivono come sono codificati i record. L’esempio usa un punto e virgola in un campo tra virgolette.\n\n## 13. L’estensione .csv non identifica il separatore\n\nUn file .csv può contenere campi separati da virgole o punti e virgola. Usa il contratto del produttore o esamina un campione del testo.\n\n## 14. Separatore sbagliato\n\nUn lettore impostato sulla virgola vede un solo campo nell’intestazione separata da punti e virgola. Verifica nomi e tipi delle colonne risultanti.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "spark.read.text(semicolon).show(truncate=False)\nquoted = fixture(\"quoted.csv\", 'id;note\\n1;\"tea; no sugar\"\\n')\nassert spark.read.options(header=True, sep=\";\", quote='\"', multiLine=True).csv(quoted).first().note == \"tea; no sugar\"\nassert spark.read.csv(semicolon, header=True).columns == [\"id;day;units\"]",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Modalità per i dati non validi\n\nPERMISSIVE conserva i campi recuperabili e può catturare il record errato originale. DROPMALFORMED scarta i record malformati quando li rileva. FAILFAST genera un errore quando ne incontra uno.\n\n## 16. Conservare il record non valido\n\nDichiara un campo stringa chiamato `_corrupt_record` e usalo come colonna dei record corrotti. I dati di prova contengono un errore di conversione di tipo.\n\n## 17. Perché count() può non rilevare righe malformate\n\nUn conteggio può escludere campi inutilizzati ed evitare di interpretare il valore non valido. Da solo è un’azione di convalida inaffidabile. Confrontalo con la materializzazione di tutti i campi nella modalità scelta.\n\n## 18. Interrogare solo il campo dei record corrotti\n\nPer sorgenti CSV e JSON grezze, Spark limita le query che fanno riferimento soltanto al campo dei record corrotti. Metti in cache e materializza il DataFrame letto prima di proiettare quel solo campo.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "bad_path = fixture(\"bad.csv\", \"id,units\\n1,2\\n2,wrong\\n3,4\\n\")\n\ndef read_bad(mode):\n    return spark.read.options(\n        header=True, mode=mode, columnNameOfCorruptRecord=\"_corrupt_record\",\n    ).schema(\"id INT, units INT, _corrupt_record STRING\").csv(bad_path)\n\ndropped = read_bad(\"DROPMALFORMED\")\nprint(\"Count with possible pruning:\", dropped.count())\nassert {r.id for r in dropped.collect()} == {1, 3}\ntry:\n    read_bad(\"FAILFAST\").collect()\nexcept (PySparkException, Py4JJavaError) as error:\n    assert \"MALFORMED_RECORD_IN_PARSING\" in str(error)\n    print(\"FAILFAST rejected the malformed record.\")\nelse:\n    raise AssertionError(\"Expected an invalid integer to fail.\")\n\nparsed = read_bad(\"PERMISSIVE\").cache()\ntry:\n    parsed.collect()\n    rejected = parsed.select(\"_corrupt_record\").where(\"_corrupt_record IS NOT NULL\").collect()\n    assert [row[0] for row in rejected] == [\"2,wrong\"]\nfinally:\n    parsed.unpersist()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 19. Leggere una cartella o un modello di nomi\n\nPassa una cartella per un dataset con file compatibili. Usa `pathGlobFilter` per limitare i nomi dei file senza alterare la scoperta delle partizioni.\n\n## 20. Scoperta delle partizioni\n\nSpark riconosce i nomi delle cartelle con coppie chiave-valore, come `zone=A`. Leggi la radice del dataset o fornisci `basePath` quando leggi solo alcune sottocartelle.\n\n## 21. Colonna nascosta dei metadati dei file\n\nAlcune sorgenti file Spark espongono una struct nascosta `_metadata`. Se supportato, selezionane un campo esplicitamente e controlla che nessuna colonna di input nasconda il nome dei metadati. La lezione usa `input_file_name()` di Apache Spark per mostrare la provenienza del file in Spark classico.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "fixture(\"zones/zone=A/a.csv\", \"id,units\\n1,2\\n\")\nfixture(\"zones/zone=B/b.csv\", \"id,units\\n2,3\\n\")\nfixture(\"zones/zone=B/notes.txt\", \"ignore this\")\nzones = spark.read.options(header=True, pathGlobFilter=\"*.csv\").schema(\"id INT, units INT\").csv(str(root / \"zones\"))\nassert {r.zone for r in zones.select(\"zone\").collect()} == {\"A\", \"B\"}\nzones.select(\"*\", F.input_file_name().alias(\"source_file\")).show(truncate=False)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 22. Leggere Parquet\n\nParquet conserva i tipi insieme ai dati. Scrivi un piccolo dataset locale e rileggilo. Qui `coalesce(1)` riduce le partizioni di output senza richiedere uno shuffle.\n\n## 23. Unire gli schemi dei file\n\nUsa `mergeSchema=true` per leggere insieme varianti compatibili dello schema. Il campo aggiunto vale null nelle righe più vecchie.\n\n## 24. mergeSchema con molti file\n\nUnire gli schemi richiede l’ispezione dei metadati dei file e può aumentare il costo di pianificazione. Con input di produzione stabili, gestisci esplicitamente l’evoluzione e fornisci uno schema noto quando opportuno.\n\n## 25. Leggere un file nuovo\n\nCrea una nuova lettura dopo aver aggiunto file. DataFrame e indici dei file possono riflettere quelli individuati durante la pianificazione. Ricrea o aggiorna i lettori quando cambia la sorgente.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "parquet_path = str(root / \"parquet\")\nspark.createDataFrame([(1, 2)], \"id INT, units INT\").coalesce(1).write.parquet(parquet_path)\nassert spark.read.parquet(parquet_path).schema[\"units\"].dataType.simpleString() == \"int\"\nspark.createDataFrame([(2, 3, \"web\")], \"id INT, units INT, channel STRING\").coalesce(1).write.mode(\"append\").parquet(parquet_path)\nmerged = spark.read.option(\"mergeSchema\", True).parquet(parquet_path)\nassert {r.id: r.channel for r in merged.collect()} == {1: None, 2: \"web\"}",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 26. Leggere da un database\n\nIl lettore JDBC usa un URL JDBC, una tabella e un driver compatibile sul classpath di Spark. Imposta `RUN_JDBC=1` solo dopo aver configurato database, driver e variabili d’ambiente indicate. Per impostazione predefinita Run All salta la chiamata esterna.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "if os.environ.get(\"RUN_JDBC\") == \"1\":\n    database = spark.read.format(\"jdbc\").options(\n        url=os.environ[\"DEMO_JDBC_URL\"],\n        dbtable=os.environ[\"DEMO_JDBC_TABLE\"],\n        driver=os.environ[\"DEMO_JDBC_DRIVER\"],\n        user=os.environ[\"DEMO_JDBC_USER\"],\n        password=os.environ[\"DEMO_JDBC_PASSWORD\"],\n    ).load()\n    database.printSchema()\nelse:\n    print(\"JDBC example skipped. Set RUN_JDBC=1 after configuring the database and driver.\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 27. Sorgente dati scritta in Python\n\nSpark 4 può registrare una sorgente implementata in Python. Il lettore fornisce partizioni di input ed emette righe dello schema dichiarato. L’esempio usa due descrittori di partizione e funziona senza servizi esterni.",
      "id": "cell-17"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "class DemoReader(DataSourceReader):\n    def partitions(self):\n        return [InputPartition(1), InputPartition(2)]\n\n    def read(self, partition):\n        yield (partition.value, partition.value * 10)\n\nclass DemoSource(DataSource):\n    @classmethod\n    def name(cls):\n        return \"nb15_demo\"\n\n    def schema(self):\n        return \"id INT, units INT\"\n\n    def reader(self, schema):\n        return DemoReader()\n\nspark.dataSource.register(DemoSource)\ndemo = spark.read.format(\"nb15_demo\").load()\nassert [tuple(r) for r in demo.orderBy(\"id\").collect()] == [(1, 10), (2, 20)]\ndemo.show()\nworkspace.cleanup()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-18"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché un file .csv può produrre una sola colonna?\n2. Perché il solo conteggio delle righe è insufficiente per controllare input malformati?\n3. Che cosa succede a una colonna aggiunta nei file Parquet più recenti quando si uniscono gli schemi?\n\n### Risposte\n\n1. Il separatore configurato può essere diverso da quello del file.\n2. Spark può escludere i campi che il conteggio non usa.\n3. La nuova colonna viene inclusa e vale null per le righe più vecchie che non avevano quel campo.",
      "id": "cell-19"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}