Data quality checks every pipeline should run

Bad data rarely announces itself. It sits in a table until a dashboard looks odd, a customer complains or a decision is made on the wrong number. Automated checks move that discovery to the moment the data arrives.

The simplest useful design is a list of checks where each check is a query that returns the bad rows. An empty result means it passes. That one rule makes checks easy to write, easy to read and easy to show to the person who has to fix the data.

A runnable example

This script builds two small tables with problems planted in them (a missing total, a duplicate id, a misspelt status, an order for a customer that does not exist) and runs the checks.

import sqlite3
from datetime import date

conn = sqlite3.connect(":memory:")
conn.executescript(
    """
    CREATE TABLE customers (id INTEGER PRIMARY KEY, email TEXT);
    CREATE TABLE orders (id INTEGER, customer_id INTEGER, status TEXT, total REAL, loaded_on TEXT);
    INSERT INTO customers VALUES (1, 'ana@example.com'), (2, 'ben@example.com');
    INSERT INTO orders VALUES
      (1, 1, 'paid',    40.0, '2025-03-10'),
      (2, 2, 'shipped', 15.5, '2025-03-10'),
      (3, 9, 'paid',    22.0, '2025-03-10'),
      (4, 1, 'refnded', 10.0, '2025-03-10'),
      (5, 2, 'paid',    NULL, '2025-03-10'),
      (5, 2, 'paid',    12.0, '2025-03-10');
    """
)

# each check is a query that returns the BAD rows: an empty result means the check passes
CHECKS = {
    "orders.total is never null":
        "SELECT id FROM orders WHERE total IS NULL",
    "orders.id is unique":
        "SELECT id FROM orders GROUP BY id HAVING COUNT(*) > 1",
    "orders.status is an accepted value":
        "SELECT id, status FROM orders WHERE status NOT IN ('new', 'paid', 'shipped', 'refunded', 'cancelled')",
    "every order has a customer":
        "SELECT o.id FROM orders o LEFT JOIN customers c ON c.id = o.customer_id WHERE c.id IS NULL",
    "no negative totals":
        "SELECT id FROM orders WHERE total < 0",
}

failures = 0
for name, sql in CHECKS.items():
    bad = conn.execute(sql).fetchall()
    status = "ok  " if not bad else "FAIL"
    failures += bool(bad)
    print(f"{status} {name}" + (f"  -> {bad}" if bad else ""))

# freshness and volume are checks too
latest = conn.execute("SELECT MAX(loaded_on) FROM orders").fetchone()[0]
age_days = (date(2025, 3, 12) - date.fromisoformat(latest)).days
print(f"{'ok  ' if age_days <= 1 else 'FAIL'} newest load is {age_days} day(s) old")
failures += age_days > 1

raise SystemExit(1 if failures else 0)

Its output, and a non-zero exit code that tells a scheduler the run failed:

FAIL orders.total is never null  -> [(5,)]
FAIL orders.id is unique  -> [(5,)]
FAIL orders.status is an accepted value  -> [(4, 'refnded')]
FAIL every order has a customer  -> [(3,)]
ok   no negative totals
FAIL newest load is 2 day(s) old

Each failure names the table and column and lists the offending ids, so the person who gets the alert knows where to look. The exit code matters: a scheduler or CI job can stop on it.

The checklist

The ten checks in the list above cover most real failures. The first five are about the shape of the data, and the next five about whether the pipeline is behaving.

Choosing what a failure does

Not every check deserves the same response. A useful way to classify them:

• Block: the data would corrupt downstream tables, such as duplicate keys in a table that feeds revenue. Stop the load.
• Quarantine: a handful of rows are bad but the rest are fine. Move the bad rows to a rejects table and continue.
• Warn: something is unusual but plausible, such as a volume 30% below normal. Alert and let a person decide.

Write the choice next to each check. When a check starts failing at 2 a.m., the decision has already been made calmly in daylight.

Keeping checks useful

• Start with the few that matter. Ten checks on the tables that feed decisions beat three hundred nobody reads.
• Treat a flapping check as a bug. One that fails and recovers every other day trains people to ignore alerts. Fix the data or the threshold.
• Add a check whenever data goes wrong. Every incident should leave a check behind so the same problem cannot return unnoticed.
• Put constraints in the database where you can. A NOT NULL, UNIQUE or CHECK constraint stops bad data at the door. Pipeline checks cover the rules a constraint cannot express.