{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "# 14 · Gestire valori mancanti e anomali\n\nPulisci i dati sulle consegne conservando le evidenze alla base di ogni scelta. Esegui in una nuova sessione CPU su Kaggle con PySpark classico 4.0.x e Java 17+. Se necessario, installa con `%pip install \"pyspark>=4.0,<4.1\" numpy` in una cella di preparazione. Il CSV di prova è incorporato come stringhe.",
      "id": "cell-01"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "from pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\nfrom pyspark.ml.feature import Imputer\n\nspark = SparkSession.builder.master(\"local[2]\").appName(\"14-missing-values-and-outliers\").getOrCreate()\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"2\")\ncsv_lines = [\n    \"delivery_id,zone,minutes,backup_minutes\",\n    \"1,A,10,12\", \"2,A,10,11\", \"3,A,10,12\", \"4,A,10,11\",\n    \"5,A,10,10\", \"6,A,120,12\", \"7,B,20,21\", \"8,B,21,22\",\n    \"9,B,22,23\", \"10,B,-999,22\", \"11,B,,24\", \"12,,21,\",\n]\nraw = spark.read.options(header=True, inferSchema=True).csv(spark.sparkContext.parallelize(csv_lines, 1))\nassert raw.count() == 12",
      "execution_count": null,
      "outputs": [],
      "id": "cell-02"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 1. Percentuale di null per colonna\n\nPer ogni campo, dividi il numero di null per il totale delle righe. I dati di prova non sono vuoti. Nel codice riutilizzabile, definisci esplicitamente il risultato con un input vuoto. Il lettore CSV usa la prima riga per i nomi e deduce i tipi dai valori.\n\n## 2. Sostituire i null con un valore\n\nUsa `fillna()` per una sostituzione documentata. Qui una zona sconosciuta riceve un’etichetta esplicita da mostrare.",
      "id": "cell-03"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "raw.agg(*[\n    (100 * F.avg(F.col(name).isNull().cast(\"double\"))).alias(name)\n    for name in raw.columns\n]).show()\nassert raw.fillna({\"zone\": \"unknown\"}).filter(\"zone = 'unknown'\").count() == 1",
      "execution_count": null,
      "outputs": [],
      "id": "cell-04"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 3. Tipi di mancanza\n\nUn valore può essere non registrato, indisponibile alla fonte o non applicabile. Questi casi possono richiedere trattamenti diversi. Qui la durata vuota è una misura non registrata.\n\n## 4. Null finti\n\nUna sentinella come -999 può indicare un dato mancante. Verificane il significato alla fonte prima di sostituirla.\n\n## 5. Trasformare i null finti in null\n\nSostituisci la sentinella nella colonna prevista e conserva il valore originale. La catena di metodi restituisce un nuovo DataFrame a ogni passaggio.\n\n## 6. Eliminare le righe con null\n\nUsa `dropna(subset=...)` per richiedere la presenza di campi specifici per un calcolo.\n\n## 7. dropna senza subset\n\nIl comportamento predefinito elimina la riga se manca un campo qualsiasi. Può quindi escludere durate utilizzabili solo perché manca la zona o il valore di riserva.",
      "id": "cell-05"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "clean = (\n    raw.withColumnRenamed(\"minutes\", \"raw_minutes\")\n    .withColumn(\"minutes\", F.when(F.col(\"raw_minutes\") != -999, F.col(\"raw_minutes\")).cast(\"double\"))\n)\nassert clean.filter(\"minutes IS NULL\").count() == 2\nassert clean.dropna(subset=[\"minutes\"]).count() == 10\nassert clean.dropna().count() == 9",
      "execution_count": null,
      "outputs": [],
      "id": "cell-06"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 8. Primo valore non null\n\n`coalesce()` sceglie la prima espressione presente nell’ordine di priorità indicato. Usa il dato di riserva solo se la misura è adatta.\n\n## 9. Valore più frequente\n\nConta i valori presenti per categoria e ordina per frequenza decrescente. Una chiave di ordinamento secondaria risolve esplicitamente i pari merito.\n\n## 10. Imputazione con Imputer\n\nAddestra un imputatore a mediana sulla colonna numerica ripulita. Il modello aggiunge una colonna separata con i valori imputati. Per un modello predittivo, addestra l’imputatore solo sui dati di addestramento.",
      "id": "cell-07"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "fallback = clean.withColumn(\"best_minutes\", F.coalesce(\"minutes\", \"backup_minutes\"))\nassert fallback.filter(\"delivery_id = 10\").first().best_minutes == 22\nmode = clean.dropna(subset=[\"zone\"]).groupBy(\"zone\").count().orderBy(F.desc(\"count\"), \"zone\").first()\nprint(\"Most frequent zone:\", mode.zone)\nmodel = Imputer(strategy=\"median\", inputCols=[\"minutes\"], outputCols=[\"median_filled\"]).fit(clean)\nassert model.transform(clean).filter(\"median_filled IS NULL\").count() == 0",
      "execution_count": null,
      "outputs": [],
      "id": "cell-08"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 11. Valore anomalo\n\nUn valore anomalo è insolito rispetto a una distribuzione di riferimento. La consegna di 120 minuti spicca nella zona A.\n\n## 12. La media risente dei valori anomali\n\nConfronta la media con e senza la consegna lunga. Rimuoverla cambia sensibilmente il riepilogo.\n\n## 13. Indicatore di mancanza\n\nMantieni un indicatore booleano che mostri se la durata ripulita mancava prima dell’imputazione.\n\n## 14. Null nelle aggregazioni\n\n`avg(minutes)` ignora i null. `count(minutes)` conta i valori presenti e `count(\"*\")` conta le righe.\n\n## 15. Operazioni aritmetiche con null\n\nAggiungere uno a una durata null restituisce null.\n\n## 16. Null nel raggruppamento\n\n`groupBy()` riunisce in un gruppo le chiavi null.\n\n## 17. Chiavi null nei join\n\nL’uguaglianza ordinaria non abbina chiavi null. L’uguaglianza con supporto ai null abbina due null. Usala solo se l’abbinamento rispetta la regola di business.\n\n## 18. Gestire i null\n\nScegli in base allo scopo. Un report può mantenere i valori sconosciuti e mostrare la copertura. Un modello può usare imputazione e indicatore di mancanza. Un calcolo che richiede durate misurate può limitarsi a un sottoinsieme.",
      "id": "cell-09"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "zone_a = clean.filter(\"zone = 'A'\")\nassert zone_a.agg(F.avg(\"minutes\")).first()[0] > zone_a.filter(\"minutes < 120\").agg(F.avg(\"minutes\")).first()[0]\nflagged = clean.withColumn(\"missing_minutes\", F.col(\"minutes\").isNull())\nflagged.groupBy(\"zone\").agg(F.count(\"*\").alias(\"rows\"), F.count(\"minutes\").alias(\"present\")).show()\nassert flagged.select((F.col(\"minutes\") + 1).alias(\"plus_one\")).filter(\"plus_one IS NULL\").count() == 2\n\nlookup = spark.createDataFrame([(\"A\",), (\"B\",), (None,)], \"zone STRING\")\nassert clean.join(lookup, \"zone\").count() == 11\nassert clean.alias(\"l\").join(lookup.alias(\"r\"), F.col(\"l.zone\").eqNullSafe(F.col(\"r.zone\"))).count() == 12",
      "execution_count": null,
      "outputs": [],
      "id": "cell-10"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## 19. Scarto interquartile\n\nL’IQR è Q3 meno Q1. Una regola comune segnala i punti fuori dall’intervallo tra Q1 meno 1,5 IQR e Q3 più 1,5 IQR.\n\n## 20. approxQuantile con errore zero\n\nUn errore zero calcola quantili esatti basati sul rango e può essere costoso su larga scala. I null vengono ignorati.\n\n## 21. IQR con dati concentrati\n\nNella zona A entrambi i quartili valgono 10, quindi l’IQR è zero ed entrambi i limiti coincidono con 10. Esamina il caso prima di trasformare la segnalazione in una regola di eliminazione.\n\n## 22. Z-score\n\nSottrai la media e dividi per la deviazione standard. Usa la divisione sicura se la deviazione standard può essere zero. Qui si usa la deviazione standard della popolazione.\n\n## 23. Effetto maschera\n\nUn valore estremo può aumentare sia la media sia la deviazione standard. La consegna di 120 minuti sposta il riferimento della zona A abbastanza da avere un proprio z-score assoluto inferiore a tre.\n\n## 24. Limiti per gruppo\n\nI riferimenti per le consegne possono variare per zona. Calcola i quartili per zona e mostra la dimensione dei gruppi. Questi piccoli gruppi servono da esempio.\n\n## 25. Errore o evento reale\n\nUna durata di 120 minuti potrebbe essere un ritardo reale o un errore di registrazione. Verifica le evidenze alla fonte prima di cambiarla.\n\n## 26. Segnalare i valori anomali\n\nConserva il valore originale e aggiungi un indicatore. Qui l’indicatore resta null quando la misura o i limiti del gruppo sono sconosciuti.\n\n## 27. Limitare i valori cambia i totali\n\nImporre una soglia massima riduce il totale dei minuti misurati nell’esempio. Conserva i valori limitati in una colonna separata e misura la differenza.\n\n## 28. Asimmetria\n\nL’asimmetria positiva indica una coda verso destra. Qui la consegna lunga allunga la coda destra della zona A.\n\n## 29. Scala logaritmica\n\n`log1p()` comprime i valori non negativi più grandi preservandone l’ordine. Il risultato è su una scala trasformata e non rappresenta più minuti.\n\n## 30. Gestire i valori anomali\n\nMantieni i ritardi lunghi verificati per l’analisi operativa. Correggi gli errori di registrazione confermati usando le evidenze alla fonte. Se un modello usa un limite massimo, documentalo e valutane l’effetto.",
      "id": "cell-11"
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": "q1, q3 = zone_a.approxQuantile(\"minutes\", [0.25, 0.75], 0)\nassert (q1, q3) == (10.0, 10.0)\nstats = zone_a.agg(F.avg(\"minutes\").alias(\"mean\"), F.stddev_pop(\"minutes\").alias(\"sd\")).first()\nz_score = (F.col(\"minutes\") - F.lit(stats[\"mean\"]))\nz_score = F.try_divide(z_score, F.lit(stats.sd))\nassert 0 < zone_a.filter(\"minutes = 120\").select(z_score).first()[0] < 3\n\nbounds = clean.filter(\"zone IS NOT NULL\").groupBy(\"zone\").agg(\n    F.count(\"minutes\").alias(\"sample_size\"),\n    F.percentile_approx(\"minutes\", [0.25, 0.75]).alias(\"quartiles\"),\n)\nscored = clean.join(bounds, \"zone\", \"left\")\nlower = F.col(\"quartiles\")[0]\nupper = F.col(\"quartiles\")[1]\niqr = upper - lower\nscored = scored.withColumn(\n    \"outlier\", (F.col(\"minutes\") < lower - 1.5 * iqr) | (F.col(\"minutes\") > upper + 1.5 * iqr)\n)\nassert scored.filter(\"zone = 'A' AND outlier\").first().delivery_id == 6\nscored.select(\"delivery_id\", \"zone\", \"minutes\", \"sample_size\", \"outlier\").orderBy(\"delivery_id\").show()\n\ncapped = clean.withColumn(\"capped_minutes\", F.when(F.col(\"minutes\").isNotNull(), F.least(\"minutes\", F.lit(60.0))))\nassert capped.agg(F.sum(\"minutes\") - F.sum(\"capped_minutes\")).first()[0] == 60\nassert zone_a.agg(F.skewness(\"minutes\")).first()[0] > 0\nzone_a.select(\"minutes\", F.log1p(\"minutes\").alias(\"log_minutes\")).show()",
      "execution_count": null,
      "outputs": [],
      "id": "cell-12"
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": "## Verifica la comprensione\n\n1. Perché i dati ripuliti conservano i minuti originali?\n2. Perché uno z-score può non segnalare la consegna più lunga?\n3. Quale evidenza giustificherebbe modificare la durata originale?\n\n### Risposte\n\n1. Per conservare la distinzione tra una sentinella alla fonte e un input vuoto.\n2. La consegna lunga aumenta la media e la deviazione standard di riferimento usate per valutarla.\n3. Una registrazione alla fonte o una correzione verificata può stabilire che il valore originale è errato.",
      "id": "cell-13"
    }
  ],
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 5
}