14 · Gestire valori mancanti e anomali

Pulisci 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.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.ml.feature import Imputer

spark = SparkSession.builder.master("local[2]").appName("14-missing-values-and-outliers").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "2")
csv_lines = [
    "delivery_id,zone,minutes,backup_minutes",
    "1,A,10,12", "2,A,10,11", "3,A,10,12", "4,A,10,11",
    "5,A,10,10", "6,A,120,12", "7,B,20,21", "8,B,21,22",
    "9,B,22,23", "10,B,-999,22", "11,B,,24", "12,,21,",
]
raw = spark.read.options(header=True, inferSchema=True).csv(spark.sparkContext.parallelize(csv_lines, 1))
assert raw.count() == 12

1. Percentuale di null per colonna

Per 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.

2. Sostituire i null con un valore

Usa fillna() per una sostituzione documentata. Qui una zona sconosciuta riceve un’etichetta esplicita da mostrare.

raw.agg(*[
    (100 * F.avg(F.col(name).isNull().cast("double"))).alias(name)
    for name in raw.columns
]).show()
assert raw.fillna({"zone": "unknown"}).filter("zone = 'unknown'").count() == 1

3. Tipi di mancanza

Un 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.

4. Null finti

Una sentinella come -999 può indicare un dato mancante. Verificane il significato alla fonte prima di sostituirla.

5. Trasformare i null finti in null

Sostituisci la sentinella nella colonna prevista e conserva il valore originale. La catena di metodi restituisce un nuovo DataFrame a ogni passaggio.

6. Eliminare le righe con null

Usa dropna(subset=...) per richiedere la presenza di campi specifici per un calcolo.

7. dropna senza subset

Il 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.

clean = (
    raw.withColumnRenamed("minutes", "raw_minutes")
    .withColumn("minutes", F.when(F.col("raw_minutes") != -999, F.col("raw_minutes")).cast("double"))
)
assert clean.filter("minutes IS NULL").count() == 2
assert clean.dropna(subset=["minutes"]).count() == 10
assert clean.dropna().count() == 9

8. Primo valore non null

coalesce() sceglie la prima espressione presente nell’ordine di priorità indicato. Usa il dato di riserva solo se la misura è adatta.

9. Valore più frequente

Conta i valori presenti per categoria e ordina per frequenza decrescente. Una chiave di ordinamento secondaria risolve esplicitamente i pari merito.

10. Imputazione con Imputer

Addestra 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.

fallback = clean.withColumn("best_minutes", F.coalesce("minutes", "backup_minutes"))
assert fallback.filter("delivery_id = 10").first().best_minutes == 22
mode = clean.dropna(subset=["zone"]).groupBy("zone").count().orderBy(F.desc("count"), "zone").first()
print("Most frequent zone:", mode.zone)
model = Imputer(strategy="median", inputCols=["minutes"], outputCols=["median_filled"]).fit(clean)
assert model.transform(clean).filter("median_filled IS NULL").count() == 0

11. Valore anomalo

Un valore anomalo è insolito rispetto a una distribuzione di riferimento. La consegna di 120 minuti spicca nella zona A.

12. La media risente dei valori anomali

Confronta la media con e senza la consegna lunga. Rimuoverla cambia sensibilmente il riepilogo.

13. Indicatore di mancanza

Mantieni un indicatore booleano che mostri se la durata ripulita mancava prima dell’imputazione.

14. Null nelle aggregazioni

avg(minutes) ignora i null. count(minutes) conta i valori presenti e count("*") conta le righe.

15. Operazioni aritmetiche con null

Aggiungere uno a una durata null restituisce null.

16. Null nel raggruppamento

groupBy() riunisce in un gruppo le chiavi null.

17. Chiavi null nei join

L’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.

18. Gestire i null

Scegli 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.

zone_a = clean.filter("zone = 'A'")
assert zone_a.agg(F.avg("minutes")).first()[0] > zone_a.filter("minutes < 120").agg(F.avg("minutes")).first()[0]
flagged = clean.withColumn("missing_minutes", F.col("minutes").isNull())
flagged.groupBy("zone").agg(F.count("*").alias("rows"), F.count("minutes").alias("present")).show()
assert flagged.select((F.col("minutes") + 1).alias("plus_one")).filter("plus_one IS NULL").count() == 2

lookup = spark.createDataFrame([("A",), ("B",), (None,)], "zone STRING")
assert clean.join(lookup, "zone").count() == 11
assert clean.alias("l").join(lookup.alias("r"), F.col("l.zone").eqNullSafe(F.col("r.zone"))).count() == 12

19. Scarto interquartile

L’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.

20. approxQuantile con errore zero

Un errore zero calcola quantili esatti basati sul rango e può essere costoso su larga scala. I null vengono ignorati.

21. IQR con dati concentrati

Nella 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.

22. Z-score

Sottrai 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.

23. Effetto maschera

Un 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.

24. Limiti per gruppo

I 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.

25. Errore o evento reale

Una durata di 120 minuti potrebbe essere un ritardo reale o un errore di registrazione. Verifica le evidenze alla fonte prima di cambiarla.

26. Segnalare i valori anomali

Conserva il valore originale e aggiungi un indicatore. Qui l’indicatore resta null quando la misura o i limiti del gruppo sono sconosciuti.

27. Limitare i valori cambia i totali

Imporre una soglia massima riduce il totale dei minuti misurati nell’esempio. Conserva i valori limitati in una colonna separata e misura la differenza.

28. Asimmetria

L’asimmetria positiva indica una coda verso destra. Qui la consegna lunga allunga la coda destra della zona A.

29. Scala logaritmica

log1p() comprime i valori non negativi più grandi preservandone l’ordine. Il risultato è su una scala trasformata e non rappresenta più minuti.

30. Gestire i valori anomali

Mantieni 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.

q1, q3 = zone_a.approxQuantile("minutes", [0.25, 0.75], 0)
assert (q1, q3) == (10.0, 10.0)
stats = zone_a.agg(F.avg("minutes").alias("mean"), F.stddev_pop("minutes").alias("sd")).first()
z_score = (F.col("minutes") - F.lit(stats["mean"]))
z_score = F.try_divide(z_score, F.lit(stats.sd))
assert 0 < zone_a.filter("minutes = 120").select(z_score).first()[0] < 3

bounds = clean.filter("zone IS NOT NULL").groupBy("zone").agg(
    F.count("minutes").alias("sample_size"),
    F.percentile_approx("minutes", [0.25, 0.75]).alias("quartiles"),
)
scored = clean.join(bounds, "zone", "left")
lower = F.col("quartiles")[0]
upper = F.col("quartiles")[1]
iqr = upper - lower
scored = scored.withColumn(
    "outlier", (F.col("minutes") < lower - 1.5 * iqr) | (F.col("minutes") > upper + 1.5 * iqr)
)
assert scored.filter("zone = 'A' AND outlier").first().delivery_id == 6
scored.select("delivery_id", "zone", "minutes", "sample_size", "outlier").orderBy("delivery_id").show()

capped = clean.withColumn("capped_minutes", F.when(F.col("minutes").isNotNull(), F.least("minutes", F.lit(60.0))))
assert capped.agg(F.sum("minutes") - F.sum("capped_minutes")).first()[0] == 60
assert zone_a.agg(F.skewness("minutes")).first()[0] > 0
zone_a.select("minutes", F.log1p("minutes").alias("log_minutes")).show()

Verifica la comprensione

1. Perché i dati ripuliti conservano i minuti originali?
2. Perché uno z-score può non segnalare la consegna più lunga?
3. Quale evidenza giustificherebbe modificare la durata originale?

Risposte

1. Per conservare la distinzione tra una sentinella alla fonte e un input vuoto.
2. La consegna lunga aumenta la media e la deviazione standard di riferimento usate per valutarla.
3. Una registrazione alla fonte o una correzione verificata può stabilire che il valore originale è errato.