19 · Verificare e diagnosticare il codice PySpark

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

from pyspark.errors import AnalysisException, PySparkException, PythonException
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql import functions as F
from pyspark.testing.utils import assertDataFrameEqual, assertSchemaEqual

spark = SparkSession.builder.master("local[2]").appName("19-test-and-debug").getOrCreate()
spark.conf.set("spark.sql.ansi.enabled", "true")
schema = "order_id INT, item STRING, quantity INT"
incoming = spark.createDataFrame(
    [(1, " Tea ", 2), (2, "Cake", -1), (3, "Coffee", None), (4, "Tea", 0)], schema
)

1. Funzione da DataFrame a DataFrame

Una funzione di trasformazione riceve un DataFrame e ne restituisce uno. Qui preserva l’input e restituisce etichette ripulite.

2. Catena di funzioni

transform() applica una funzione DataFrame e passa il risultato al passo successivo.

3. Strutturare il codice

Separa le trasformazioni da input, output e azioni. Documenta colonne richieste e unità di misura. Lascia al job chiamante la scelta tra rifiuto e quarantena.

def normalize_items(df: DataFrame) -> DataFrame:
    return df.withColumn("item", F.lower(F.trim("item")))

def add_quantity_rule(df: DataFrame) -> DataFrame:
    quantity = F.col("quantity")
    return df.withColumn("quantity_valid", F.coalesce(quantity >= 0, F.lit(False)))

checked = incoming.transform(normalize_items).transform(add_quantity_rule)
checked.orderBy("order_id").show()

4. Metrica di qualità

Conta righe e violazioni, poi calcola il tasso di violazione. Per un input vuoto il tasso è sconosciuto.

5. Regola di qualità

La quantità deve essere presente e non negativa. Qui zero è valido.

6. Regola con un null

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

valid = F.col("quantity_valid")
metrics = checked.agg(
    F.count("*").alias("rows"),
    F.count(F.when(~valid, 1)).alias("violations"),
).withColumn("violation_rate", F.try_divide("violations", "rows"))
quality = metrics.first()
assert (quality.rows, quality.violations, quality.violation_rate) == (4, 2, 0.5)
assert incoming.filter(~(F.col("quantity") >= 0)).count() == 1

7. Fermare il job su una violazione

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

8. Quarantena

Una politica di quarantena separa le righe rifiutate e aggiunge un motivo. Conserva gli identificativi e i valori originali per esaminarli e correggerli.

def require_clean_batch(violation_count: int) -> None:
    if violation_count:
        raise ValueError(f"Batch rejected: {violation_count} invalid quantities")

try:
    require_clean_batch(quality.violations)
except ValueError as error:
    assert str(error) == "Batch rejected: 2 invalid quantities"
    print(error)
else:
    raise AssertionError("Expected the quality gate to reject the batch.")

reason = F.when(F.col("quantity").isNull(), "missing_quantity").otherwise("negative_quantity")
quarantine = incoming.transform(add_quantity_rule).filter(~valid).withColumn("reason", reason)
accepted = checked.filter(valid)
assert accepted.count() + quarantine.count() == quality.rows
quarantine.orderBy("order_id").show()

9. Test unitario

Un test unitario verifica un contratto specifico con input noti e risultati attesi. Qui verifica insieme le due trasformazioni.

10. Qualità dei dati e test unitari

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

11. Dati di test

Includi una riga normale, un valore al limite, uno mancante e uno non valido. Usa tipi espliciti e pochi dati.

12. Confrontare gli schemi in un test

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

13. Test con dati reali

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

def test_transformations(session):
    fixture = session.createDataFrame(
        [(1, " Tea ", 2), (2, "Cake", -1), (3, None, None), (4, "Tea", 0)], schema
    )
    expected = session.createDataFrame(
        [(1, "tea", 2, True), (2, "cake", -1, False), (3, None, None, False), (4, "tea", 0, True)],
        schema + ", quantity_valid BOOLEAN NOT NULL",
    )
    actual = fixture.transform(normalize_items).transform(add_quantity_rule)
    assertSchemaEqual(actual.schema, expected.schema, ignoreNullable=False)
    assertDataFrameEqual(actual, expected)

test_transformations(spark)

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory(prefix="nb19-") as directory:
    csv_path = Path(directory) / "orders.csv"
    csv_path.write_text("order_id,item,quantity\n1,Tea,2\n2,Cake,\n", encoding="utf-8")
    sample = spark.read.option("header", True).schema(schema).csv(str(csv_path))
    assert sample.transform(add_quantity_rule).filter(~valid).count() == 1

14. Fixture pytest per la sessione

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

import pytest

@pytest.fixture(scope="session")
def spark_test_session():
    session = SparkSession.builder.master("local[2]").appName("pytest-nb19").getOrCreate()
    try:
        yield session
    finally:
        session.stop()

def test_order_transformations(spark_test_session):
    test_transformations(spark_test_session)

15. Errore di analisi

Un riferimento a una colonna assente fallisce durante la risoluzione. Spark classico può segnalare l’errore già durante la costruzione del DataFrame.

16. Errore all’azione

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

17. Classe di errore

In Spark 4, getCondition() espone un identificativo strutturato per gli errori PySpark. Usalo con il messaggio e il contesto della query per la diagnosi.

try:
    incoming.select("missing_column").collect()
except AnalysisException as error:
    assert error.getCondition().startswith("UNRESOLVED_COLUMN")
    print("Analysis:", error.getCondition())
else:
    raise AssertionError("Expected the missing column to fail.")

text_input = spark.createDataFrame([("bad",)], "quantity_text STRING")
casted = text_input.select(F.col("quantity_text").cast("int").alias("quantity"))
try:
    casted.collect()
except PySparkException as error:
    assert error.getCondition() == "CAST_INVALID_INPUT"
    print("Execution:", error.getCondition())
else:
    raise AssertionError("Expected malformed text to fail.")

18. Errore nel codice Python

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

@F.udf("int", useArrow=False)
def parse_quantity(value):
    return int(value)

try:
    text_input.select(parse_quantity("quantity_text").alias("quantity")).collect()
except PythonException as error:
    assert "ValueError" in str(error)
    print("Python worker: ValueError while parsing quantity.")
else:
    raise AssertionError("Expected the Python UDF to fail.")

Verifica la comprensione

1. Perché filtrare sulla negazione di un confronto che può dare null può perdere dati non validi?
2. Quale caso di prova verifica il limite della non negatività?
3. Dove cerchi un’eccezione generata da una UDF Python?

Risposte

1. La negazione di sconosciuto resta sconosciuto e un filtro conserva solo vero.
2. Quantità zero.
3. Leggi il traceback del worker Python nell’eccezione segnalata.