How would you perform data validation and quality checks in PySpark?

Instruction: Outline the methods to ensure data quality and integrity in PySpark processing pipelines.

Context: This question tests the candidate's approach to maintaining high data quality in PySpark applications, including techniques for validation and anomaly detection.

Updated

Define a data contract: required columns/types, units, key uniqueness, allowed ranges, relationships, freshness and completeness. Parsing or casting successfully does not prove that the values satisfy that contract. A failed cast can raise or yield null depending on the operation and ANSI settings; preserve the original value and handle that policy explicitly.

For an existing session named spark, this complete example keeps every input row while marking duplicate IDs and an illustrative amount rule:

from pyspark.sql import Window, functions as F

raw = spark.createDataFrame([
    (1, 10.0), (1, 12.0), (2, -2.0), (3, None), (4, 0.0), (5, float("nan"))
], "id int, amount double")
checked = raw.withColumn("duplicate_id",
    F.col("id").isNotNull() & (F.count("*").over(Window.partitionBy("id")) > 1))
amount_valid = (F.col("amount").isNotNull() & ~F.isnan("amount") &
                F.col("amount").between(0.0, 1_000_000.0))
checked = checked.withColumn("amount_valid", amount_valid).withColumn(
    "valid", F.col("id").isNotNull() & ~F.col("duplicate_id") & F.col("amount_valid"))
summary = checked.agg(F.count("*").alias("rows"),
    F.sum(F.when(F.col("valid"), 1).otherwise(0)).alias("valid_rows"),
    F.sum(F.when(F.col("duplicate_id"), 1).otherwise(0)).alias("duplicate_rows")).first()
assert (summary.rows, summary.valid_rows, summary.duplicate_rows) == (6, 1, 2)
assert checked.filter("valid").first().id == 4
quarantine = checked.filter(~F.col("valid"))
assert quarantine.count() == 5

The numeric domain is an explicit example rule, not a universal financial rule. Null and NaN are distinct, and infinity fails this bounded range. Duplicate flags retain both occurrences; silently keeping an arbitrary row would hide a reconciliation decision.

Validate foreign keys with a suitable anti-join, dates against a documented ordering/timezone contract and source/output counts against expected grain. Materialize checks before publishing; lazy transformations alone do not execute validation. Production gates should raise explicit errors rather than depend on Python assertions, which can be disabled.

Publish aggregate reason counts and controlled samples without exposing sensitive rows. Quarantine with provenance, rule version and retry/correction ownership; do not impute an identifier or delete an unusual value merely to make a check pass. Anomaly flags require investigation and are not proof of corruption.

Null semantics, ANSI behavior

Related Questions