Fail the run, not the row

A warehouse built on raw retail feeds silently mis-attributes revenue unless the bad data is caught, quarantined and reported. The feeds in this project are deliberately hostile: month-partitioned JSON with malformed lines, duplicate sale IDs, null foreign keys and two timestamp formats.

Read with a schema, never infer

Every source is read with an explicit StructType in PERMISSIVE mode, so a corrupt line does not kill the job — it is captured in _corrupt_record and counted. Schema drift is detected on a sample and logged before it turns into a column of nulls. In the reference run, 981 corrupt lines were captured this way.

Deduplicate deterministically

Duplicates are removed with row_number() over a total ordering, so re-running the job on the same input keeps the same row every time. 9,993 duplicate sale lines were dropped, and a second run would drop exactly the same ones.

Eight rules, one gate

Data-quality rules are declarative — a name and a condition — and produce a per-rule breakdown plus a quarantine table. The job does not fail because a row is wrong; it fails when the quarantined share crosses a threshold. That distinction is the whole point: one bad supplier file should not block the month, but a broken upstream export should.

RULES = [
    ("non_null_customer", F.col("customer_id").isNotNull()),
    ("positive_quantity", F.col("quantity") > 0),
    ("known_product", F.col("product_id").isin(known_products)),
    # …five more
]

def gate(df: DataFrame) -> tuple[DataFrame, DataFrame]:
    bad = None
    for name, cond in RULES:
        failed = df.filter(~cond).withColumn("rule", F.lit(name))
        bad = failed if bad is None else bad.unionByName(failed)
    good = df.filter(reduce(lambda a, b: a & b, [c for _, c in RULES]))
    return good, bad  # per-rule counts come from bad.groupBy("rule").count()

Model history, not just state

Dimensions carry SCD Type 2 validity windows and the fact table joins on the version that was valid at order time, so a customer’s segment change never rewrites last quarter. The fact is Hive-partitioned by year and month so Hive, Athena, Synapse or Databricks can query it in place. Marts use LAG, DENSE_RANK and cumulative windows; the MongoDB serving loads are idempotent upserts.

1,009,989 raw lines read, 50.9 s end-to-end (~19.8K rows/s), 1.30 % quarantined with a per-rule breakdown, five tests, and a JSON run-metrics file for every execution.