Data quality monitoring: record metrics on every run and alert on what changed

Part 2 stopped files that break the agreed rules. The worst problems follow the rules. A till that lost its connection at noon sends a perfectly valid file with half a day's orders. Every row passes; the day's sales are wrong. Catching that needs to know what a normal day looks like, which means keeping history. This part uses module 10's orders_daily after a backfill of September. To rebuild from nothing, run python make_days.py and module 10's python backfill.py 2026-09-02 2026-10-01.

A few metrics, on every run

Data QualityData QualityHow far data can be trusted: complete, correct, on time and in the expected shape. Pipelines check it with rules and stop or flag rows that fail. has many sides. For a daily load, 3 metrics catch most problems:

• Rows: how many orders the day has. None means the day is missing; a drop, that data went missing; a jump, often, that it was loaded twice.
• Empty customers: the share of orders with no customer_id. A rise usually means a change upstream.
• Sales: the day's total amount, the number the business reads.

Record them in a table of their own after every load, so the pipeline keeps its own history. Save this as quality.py:

"""Record a run's quality metrics in quality_metrics, then compare the day with the 7 days before it.
Exits 1 when the day looks wrong."""
import sys
from datetime import date, timedelta

from pyspark.sql import functions as F
from delta_session import delta_spark

run_date = sys.argv[1]
day = str(date.fromisoformat(run_date) - timedelta(days=1))
spark = delta_spark("quality")

metrics = (
    spark.table("orders_daily").where(F.col("order_date") == day)
    .agg(
        F.count("*").alias("orders"),
        F.avg(F.col("customer_id").isNull().cast("double")).alias("empty_customer_share"),
        F.sum("amount").alias("sales"),
    )
    .select(F.lit(day).cast("date").alias("order_date"), "orders", "empty_customer_share", "sales")
)
# Replace this day's metrics, so a rerun of the day records it once.
metrics.write.format("delta").mode("overwrite").option("replaceWhere", f"order_date = '{day}'").saveAsTable("quality_metrics")

today = spark.table("quality_metrics").where(F.col("order_date") == day).first()
week = spark.table("quality_metrics").where(
    F.col("order_date").between(F.date_sub(F.lit(day), 7), F.date_sub(F.lit(day), 1))
)
usual = week.agg(F.percentile_approx("orders", 0.5).alias("orders"), F.count("*").alias("days")).first()

problems = []
if today.orders == 0:
    problems.append("no orders: the day is missing")
elif usual.days >= 5 and today.orders < 0.5 * usual.orders:
    problems.append(f"{today.orders} orders, under half the usual {usual.orders}")
if today.empty_customer_share and today.empty_customer_share > 0.01:
    problems.append(f"{today.empty_customer_share:.0%} of orders have no customer")

for problem in problems:
    print(f"ALERT {day}: {problem}")
if not problems:
    print(f"ok {day}: {today.orders} orders, sales {today.sales}, usual {usual.orders} orders a day")
sys.exit(1 if problems else 0)

What the comparison does

The usual day is the median of the 7 days before, from percentile_approx, so 1 odd day in the week does not move it much. The check alerts when a day has under half of it. It waits until there are 5 days of history, because 2 days are not enough to know what usual is; until then it only checks for no orders at all. The empty customer share alerts above 1%, a rule rather than a comparison, because the right share is 0.

The thresholds are choices, not facts. Half the usual is loose enough that a quiet Monday does not wake anyone, and tight enough that half a day's orders does. A shop with a busy weekend would compare each day with the same weekday before it. Start loose, and tighten a threshold after it has missed something, not before: a check that alerts every week is soon ignored.

The metrics are written with the same replaceWhere as the load, so a rerun or a backfill of a day records it once, and quality_metrics is as safe to rebuild as orders_daily. In a job, quality.py runs after each day's load. To give it a history now, run it for the runs of 2 to 25 September, which record 1 to 24 September:

for n in $(seq -w 2 25); do python quality.py 2026-09-$n; done

Try it on a bad day

Cut the file for 25 September to its first order, as a till that lost its connection would send it, then load and check that day:

head -2 landing/orders/2026-09-25.csv > cut.csv && mv cut.csv landing/orders/2026-09-25.csv
python load_day.py 2026-09-26
python quality.py 2026-09-26

The cut file's 1 row is valid, so part 2's contract passes it and the load succeeds. Every day of the sample has 3 to 6 orders, so the usual day of the week before has at least 3, and quality.py prints an ALERT line for 1 order, under half of it, and exits with 1. Put the file back with python make_days.py, load the day again and the same check prints ok.

Data quality monitoring on Databricks

Databricks watches tables for some of this without code. Anomaly detection, turned on for a schema, learns from each table's history:

• Freshness: when the table is usually updated, from its history of commits. A commit that is unusually late marks the table as stale.
• Completeness: how many rows are usually written in 24 hours. A day below the lower end of the predicted range marks the table as incomplete.

Data profiling, formerly Lakehouse Monitoring, computes summary statistics of a table over time, such as the share of nulls or a column's distribution, compares windows of time and can alert on a change. Neither changes the tables it watches or slows the jobs that write them. Both run on serverless compute and are billed on their own line, by the number and size of the tables.

They are a good net across a whole schema, for tables nobody wrote checks for. Your own checks still earn their place for the rules only you know: that this table should have no empty customers, that a day's sales should match the till's total. In a Spark Declarative Pipeline the same rules are expectations, written beside the table they guard.

The job so far

For each day: the contract check, the load, check_day.py against the file, then quality.py against history. A failure at any step stops the day and says why. Part 4 puts this job, its code and its settings in 1 folder that deploys the same way every time.