Data quality, testing and observability

A pipeline that runs without errors can still deliver wrong data. A job can finish green while a source quietly sent half the rows, a column changed meaning, or the same order was loaded twice. The dashboard looks fine, and a manager makes a decision on bad numbers.

In this tutorial you will write a small data quality checker in Python and DuckDB, run it against a table with planted problems, fix the data, and learn how the same ideas appear in dbt, Great Expectations and data observability tools.

Prerequisites: basic SQL and Python 3.10 or newer. Install DuckDB with pip install duckdb. No account or server is needed.

Data quality problems are not rare events. They are the normal state of data that comes from other people's systems. Some common causes:

- A source app has a bug or a new release and sends nulls, duplicates or strange values. - A retry or a re-run loads the same file twice. - A source is late or silent, so yesterday's report shows only a few hours of orders. - Someone renames or retypes a column upstream and your pipeline breaks, or worse, keeps going with wrong values. - A join drops or multiplies rows without any error.

The first four are checks on the data itself. Consistency compares tables. Timeliness is about time. Most teams start with the cheap ones: not null, unique, accepted values, row counts and freshness. They catch the majority of real incidents.

Check data at every boundary, not only at the end. The earlier a problem is caught, the cheaper it is to fix, and the fewer people have seen the wrong numbers.

- At ingestion: is the file there, is it the expected size, does it have the expected columns? - After transformation: are keys unique, are values valid, do totals match? - Before publishing: the last gate before a dashboard or ML model reads the table.

A powerful pattern is Write, Audit, Publish (WAP). Write new data to a staging table. Audit it with your checks. Only if it passes, publish it, by swapping it into the real table or loading it into the reporting layer. If it fails, users keep seeing the last good data instead of bad data. The project lesson *Project: build an end-to-end data pipeline* uses this idea.

We will make a tiny orders table with five problems on purpose: a duplicate, a missing name, an invalid status, a negative total and (later) too few rows. Create a folder quality-demo, open a terminal inside it, and save this file.

import duckdb

con = duckdb.connect("quality.duckdb")
con.execute("""
    CREATE OR REPLACE TABLE orders (
        order_id VARCHAR, customer_name VARCHAR, status VARCHAR,
        total INTEGER, created_at TIMESTAMP
    )""")
con.execute("""
    INSERT INTO orders VALUES
    ('A-001', 'Asha',  'picked_up', 330, now() - INTERVAL 3 HOUR),
    ('A-002', 'Ravi',  'picked_up', 120, now() - INTERVAL 2 HOUR),
    ('A-003', 'Meera', 'ready',     180, now() - INTERVAL 1 HOUR),
    ('A-003', 'Meera', 'ready',     180, now() - INTERVAL 1 HOUR),
    ('A-004', NULL,    'picked_up', 90,  now() - INTERVAL 30 MINUTE),
    ('A-005', 'Imran', 'delivered', 220, now() - INTERVAL 20 MINUTE),
    ('A-006', 'Kavya', 'picked_up', -150, now() - INTERVAL 10 MINUTE)
""")
print("Created quality.duckdb with 7 orders")

Creates quality.duckdb with 7 orders. Timestamps are relative to now, so freshness checks work whenever you run it.

The idea is simple and used by many real tools: a check is a query that returns the bad rows. If it returns nothing, the check passes. This keeps checks easy to read, easy to add and easy to debug, because the failing rows are right there.

Save this as quality.py.

import sys

import duckdb

# Each check is a name plus a SQL query that returns the BAD rows.
# A check passes when the query returns zero rows.
CHECKS = [
    ("order_id is unique",
     "SELECT order_id FROM orders GROUP BY order_id HAVING count(*) > 1"),
    ("customer_name is not null",
     "SELECT order_id FROM orders WHERE customer_name IS NULL"),
    ("status is an allowed value",
     """SELECT order_id, status FROM orders
        WHERE status NOT IN ('received', 'preparing', 'ready', 'picked_up', 'cancelled')"""),
    ("total is positive",
     "SELECT order_id, total FROM orders WHERE total <= 0"),
    ("data is fresh (newest order under 2 hours old)",
     """SELECT max(created_at) AS newest FROM orders
        HAVING max(created_at) < now() - INTERVAL 2 HOUR"""),
    ("volume is at least 5 rows",
     "SELECT count(*) AS n FROM orders HAVING count(*) < 5"),
]


def run_checks(con) -> int:
    failures = 0
    for name, sql in CHECKS:
        bad = con.execute(sql).fetchall()
        if bad:
            failures += 1
            print(f"FAIL  {name}  ({len(bad)} bad)  e.g. {bad[:2]}")
        else:
            print(f"pass  {name}")
    return failures


if __name__ == "__main__":
    con = duckdb.connect("quality.duckdb", read_only=True)
    failed = run_checks(con)
    print(f"\n{failed} of {len(CHECKS)} checks failed")
    sys.exit(1 if failed else 0)

Runs every check, prints pass or FAIL with example bad rows, and exits with code 1 if anything failed so a scheduler can stop the pipeline.

Notice two details. The freshness and volume checks use HAVING so they return a row only when something is wrong. And the script exits with a non-zero code on failure: that is how Airflow, cron or a CI job knows to stop and alert.

Each failure points at the exact rows. That is the difference between "something is wrong with orders" and "order A-005 has status delivered". Good checks tell you where to look.

Finding a problem is half the job. You must also decide what happens next. There are three common policies:

- Fail the pipeline: stop, alert, publish nothing. Use for problems that make results wrong (duplicate keys, broken totals). - Quarantine the rows: move bad rows to a rejects table with a reason, continue with the good ones, and alert if the reject rate is high. Use when a few bad rows are expected. - Warn only: log it and move on. Use for soft rules you are still learning about.

Most checks should have a severity: error stops the pipeline, warn only notifies. Without severity, teams get so many alerts that they ignore all of them.

You are a senior data engineer. I own this table:

[paste the CREATE TABLE statement or a list of columns with types]

It is loaded [daily / hourly] from [source]. Suggest data quality checks in these groups: completeness, uniqueness, validity, consistency with [other table], and freshness. For each check give: the SQL query that returns bad rows, whether failure should stop the pipeline or only warn, and one real-world cause that would trigger it. Keep it to the 10 most valuable checks.

What to expect: Good answers give runnable SQL that returns bad rows, and explain severity. Run the SQL yourself on sample data before trusting it.

import duckdb

con = duckdb.connect("quality.duckdb")
con.execute("""
    CREATE OR REPLACE TABLE orders AS
    SELECT DISTINCT * FROM orders
    WHERE customer_name IS NOT NULL
      AND status IN ('received', 'preparing', 'ready', 'picked_up', 'cancelled')
      AND total > 0
""")
print(con.execute("SELECT count(*) FROM orders").fetchone()[0], "rows left")

Keeps only distinct, valid orders. Run it with python fix_data.py, then python quality.py.

You rarely build the runner yourself. Real tools give you the same pattern with more features.

dbt tests are the most common for warehouse teams. In a schema.yml file you declare tests next to the columns. dbt turns each into a query that returns failing rows, just like ours. See *Transformations with dbt* for the full setup.

version: 2

models:
  - name: orders
    columns:
      - name: order_id
        tests:
          - unique
          - not_null
      - name: status
        tests:
          - accepted_values:
              values: ['received', 'preparing', 'ready', 'picked_up', 'cancelled']
      - name: customer_id
        tests:
          - relationships:
              to: ref('customers')
              field: customer_id

Four built-in dbt tests: unique, not_null, accepted_values and relationships (a foreign key check). Run them with dbt test.

Great Expectations and Soda are open-source Python-based tools for describing expectations about data, such as "column total values are between 1 and 50000", and running them against files, databases or DataFrames. Both can produce a report and plug into Airflow. Their APIs change between versions, so follow the official docs for the version you install.

Cloud warehouses and orchestrators add their own features too, such as Airflow task failure alerts and warehouse-level constraints, but remember that many warehouses do not enforce unique or foreign key constraints; they are only hints. You must test them yourself.

Tests catch problems you thought of. Data observability is about catching problems you did not think of, by monitoring the health of data continuously. The usual five signals:

- Freshness: when did this table last update? - Volume: is the row count normal for this hour or day? - Schema: did columns appear, disappear or change type? - Distribution: did the share of nulls, or the average of total, jump suddenly? - Lineage: if this table is wrong, which dashboards and models downstream are affected, and which upstream source caused it?

Simple observability needs no vendor. Save a row count per day in a small metrics table and compare today with the average of the last seven days. If today is below half, alert. There are also commercial and open-source observability tools that learn normal ranges automatically; evaluate them only when you have enough tables that hand-written rules do not scale.

WITH daily AS (
    SELECT CAST(created_at AS DATE) AS day, count(*) AS n
    FROM orders
    GROUP BY 1
),
baseline AS (
    SELECT avg(n) AS avg_n
    FROM daily
    WHERE day < current_date AND day >= current_date - INTERVAL 7 DAY
)
SELECT d.day, d.n, b.avg_n
FROM daily d CROSS JOIN baseline b
WHERE d.day = current_date AND d.n < 0.5 * b.avg_n

Returns a row only when today's order count is less than half the 7-day average. Half is an arbitrary starting threshold; tune it to your data's natural ups and downs.

Many quality problems start upstream, where the people who change the data do not know who uses it. A data contract is an agreement between the producer team and the consumer team: the table's columns and types, allowed values, freshness promise and who to contact. Contracts are often written as YAML and checked automatically when the producer deploys, so a breaking change fails their build before it breaks your pipeline.

Every important table also needs an owner and an expectation (an SLA) such as "ready by 7 am IST". Without an owner, an alert is everyone's problem and nobody's job.

Expect questions like "How do you make sure your pipeline's data is correct?" and "A dashboard number looks wrong. How do you investigate?" Strong answers cover: checks at several stages (not only the end), idempotent loads, row-count reconciliation between source and target, quarantining bad rows, alerts with an owner, and lineage to trace a wrong number back to its source. Mentioning Write, Audit, Publish and data contracts shows you have thought beyond "I add NOT NULL".