{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 19 · Verificare e diagnosticare il codice PySpark\n\nCostruisci una trasformazione, convalida i dati in ingresso e riproduci tre tipi di errore. Usa PySpark classico 4.0.x e Java 17+. Su Kaggle, installa le dipendenze mancanti con `%pip install \"pyspark[sql]>=4.0,<4.1\" pytest` in una cella di preparazione, poi usa Run All. Gli esempi di errori attesi possono stampare diagnostica Spark.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.errors import AnalysisException, PySparkException, PythonException\nfrom pyspark.sql import DataFrame, SparkSession\nfrom pyspark.sql import functions as F\nfrom pyspark.testing.utils import assertDataFrameEqual, assertSchemaEqual\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"19-test-and-debug\").getOrCreate()\nspark.conf.set(\"spark.sql.ansi.enabled\", \"true\")\nschema = \"order_id INT, item STRING, quantity INT\"\nincoming = spark.createDataFrame(\n    [(1, \" Tea \", 2), (2, \"Cake\", -1), (3, \"Coffee\", None), (4, \"Tea\", 0)], schema\n)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Funzione da DataFrame a DataFrame\n\nUna funzione di trasformazione riceve un DataFrame e ne restituisce uno. Qui preserva l’input e restituisce etichette ripulite.\n\n## 2. Catena di funzioni\n\n`transform()` applica una funzione DataFrame e passa il risultato al passo successivo.\n\n## 3. Strutturare il codice\n\nSepara le trasformazioni da input, output e azioni. Documenta colonne richieste e unità di misura. Lascia al job chiamante la scelta tra rifiuto e quarantena.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def normalize_items(df: DataFrame) -> DataFrame:\n    return df.withColumn(\"item\", F.lower(F.trim(\"item\")))\n\ndef add_quantity_rule(df: DataFrame) -> DataFrame:\n    quantity = F.col(\"quantity\")\n    return df.withColumn(\"quantity_valid\", F.coalesce(quantity >= 0, F.lit(False)))\n\nchecked = incoming.transform(normalize_items).transform(add_quantity_rule)\nchecked.orderBy(\"order_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 4. Metrica di qualità\n\nConta righe e violazioni, poi calcola il tasso di violazione. Per un input vuoto il tasso è sconosciuto.\n\n## 5. Regola di qualità\n\nLa quantità deve essere presente e non negativa. Qui zero è valido.\n\n## 6. Regola con un null\n\n`quantity >= 0` dà null per una quantità mancante. Anche la negazione dà null, quindi filtrare soltanto sulla negazione perde quel caso. Il ripiego esplicito su falso rende falsa la regola quando manca la quantità.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "valid = F.col(\"quantity_valid\")\nmetrics = checked.agg(\n    F.count(\"*\").alias(\"rows\"),\n    F.count(F.when(~valid, 1)).alias(\"violations\"),\n).withColumn(\"violation_rate\", F.try_divide(\"violations\", \"rows\"))\nquality = metrics.first()\nassert (quality.rows, quality.violations, quality.violation_rate) == (4, 2, 0.5)\nassert incoming.filter(~(F.col(\"quantity\") >= 0)).count() == 1",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 7. Fermare il job su una violazione\n\nSe la politica rifiuta l’intero lotto, applica il controllo bloccante prima di scrivere l’output accettato. Solo in questa dimostrazione si intercetta l’eccezione attesa.\n\n## 8. Quarantena\n\nUna politica di quarantena separa le righe rifiutate e aggiunge un motivo. Conserva gli identificativi e i valori originali per esaminarli e correggerli.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def require_clean_batch(violation_count: int) -> None:\n    if violation_count:\n        raise ValueError(f\"Batch rejected: {violation_count} invalid quantities\")\n\ntry:\n    require_clean_batch(quality.violations)\nexcept ValueError as error:\n    assert str(error) == \"Batch rejected: 2 invalid quantities\"\n    print(error)\nelse:\n    raise AssertionError(\"Expected the quality gate to reject the batch.\")\n\nreason = F.when(F.col(\"quantity\").isNull(), \"missing_quantity\").otherwise(\"negative_quantity\")\nquarantine = incoming.transform(add_quantity_rule).filter(~valid).withColumn(\"reason\", reason)\naccepted = checked.filter(valid)\nassert accepted.count() + quarantine.count() == quality.rows\nquarantine.orderBy(\"order_id\").show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 9. Test unitario\n\nUn test unitario verifica un contratto specifico con input noti e risultati attesi. Qui verifica insieme le due trasformazioni.\n\n## 10. Qualità dei dati e test unitari\n\nUn test unitario verifica la logica su casi controllati. Un controllo di qualità verifica se un lotto specifico rispetta una regola. Usa lo stesso contratto quando è adatto.\n\n## 11. Dati di test\n\nIncludi una riga normale, un valore al limite, uno mancante e uno non valido. Usa tipi espliciti e pochi dati.\n\n## 12. Confrontare gli schemi in un test\n\nVerifica nomi, tipi, ordine dei campi e possibilità di null quando fanno parte del contratto. Confronta i dati separatamente. Qui l’ordine delle righe è irrilevante e tutti i valori sono esatti.\n\n## 13. Test con dati reali\n\nAggiungi un controllo di integrazione separato su un piccolo input rappresentativo e autorizzato. Usa il lettore e lo schema reali, poi verifica copertura e invarianti. Qui un CSV sintetico sostituisce i dati esterni.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "def test_transformations(session):\n    fixture = session.createDataFrame(\n        [(1, \" Tea \", 2), (2, \"Cake\", -1), (3, None, None), (4, \"Tea\", 0)], schema\n    )\n    expected = session.createDataFrame(\n        [(1, \"tea\", 2, True), (2, \"cake\", -1, False), (3, None, None, False), (4, \"tea\", 0, True)],\n        schema + \", quantity_valid BOOLEAN NOT NULL\",\n    )\n    actual = fixture.transform(normalize_items).transform(add_quantity_rule)\n    assertSchemaEqual(actual.schema, expected.schema, ignoreNullable=False)\n    assertDataFrameEqual(actual, expected)\n\ntest_transformations(spark)\n\nfrom pathlib import Path\nfrom tempfile import TemporaryDirectory\n\nwith TemporaryDirectory(prefix=\"nb19-\") as directory:\n    csv_path = Path(directory) / \"orders.csv\"\n    csv_path.write_text(\"order_id,item,quantity\\n1,Tea,2\\n2,Cake,\\n\", encoding=\"utf-8\")\n    sample = spark.read.option(\"header\", True).schema(schema).csv(str(csv_path))\n    assert sample.transform(add_quantity_rule).filter(~valid).count() == 1",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 14. Fixture pytest per la sessione\n\nMetti fixture e test in un modulo quando esegui pytest. Una fixture con ambito di sessione riusa una sessione Spark locale tra i test e la arresta alla fine. Qui la fixture viene definita senza invocarla nel kernel attivo.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "import pytest\n\n@pytest.fixture(scope=\"session\")\ndef spark_test_session():\n    session = SparkSession.builder.master(\"local[2]\").appName(\"pytest-nb19\").getOrCreate()\n    try:\n        yield session\n    finally:\n        session.stop()\n\ndef test_order_transformations(spark_test_session):\n    test_transformations(spark_test_session)",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 15. Errore di analisi\n\nUn riferimento a una colonna assente fallisce durante la risoluzione. Spark classico può segnalare l’errore già durante la costruzione del DataFrame.\n\n## 16. Errore all’azione\n\nUn’espressione può essere risolta correttamente e fallire su un valore. Qui la conversione è valida nel piano, ma il testo malformato fallisce quando il risultato viene raccolto in modalità ANSI.\n\n## 17. Classe di errore\n\nIn Spark 4, `getCondition()` espone un identificativo strutturato per gli errori PySpark. Usalo con il messaggio e il contesto della query per la diagnosi.",
      "id": "cell-13"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "try:\n    incoming.select(\"missing_column\").collect()\nexcept AnalysisException as error:\n    assert error.getCondition().startswith(\"UNRESOLVED_COLUMN\")\n    print(\"Analysis:\", error.getCondition())\nelse:\n    raise AssertionError(\"Expected the missing column to fail.\")\n\ntext_input = spark.createDataFrame([(\"bad\",)], \"quantity_text STRING\")\ncasted = text_input.select(F.col(\"quantity_text\").cast(\"int\").alias(\"quantity\"))\ntry:\n    casted.collect()\nexcept PySparkException as error:\n    assert error.getCondition() == \"CAST_INVALID_INPUT\"\n    print(\"Execution:\", error.getCondition())\nelse:\n    raise AssertionError(\"Expected malformed text to fail.\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-14"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 18. Errore nel codice Python\n\nUna UDF Python può fallire in un worker Spark. Leggi il traceback Python per trovare la funzione e l’eccezione originale. L’esempio disabilita Arrow esplicitamente e raccoglie la colonna derivata per obbligare Spark a valutarla.",
      "id": "cell-15"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "@F.udf(\"int\", useArrow=False)\ndef parse_quantity(value):\n    return int(value)\n\ntry:\n    text_input.select(parse_quantity(\"quantity_text\").alias(\"quantity\")).collect()\nexcept PythonException as error:\n    assert \"ValueError\" in str(error)\n    print(\"Python worker: ValueError while parsing quantity.\")\nelse:\n    raise AssertionError(\"Expected the Python UDF to fail.\")",
      "execution_count": null,
      "outputs": [],
      "id": "cell-16"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché filtrare sulla negazione di un confronto che può dare null può perdere dati non validi?\n2. Quale caso di prova verifica il limite della non negatività?\n3. Dove cerchi un’eccezione generata da una UDF Python?\n\n### Risposte\n\n1. La negazione di sconosciuto resta sconosciuto e un filtro conserva solo vero.\n2. Quantità zero.\n3. Leggi il traceback del worker Python nell’eccezione segnalata.",
      "id": "cell-17"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}