Data Quality: Testing and Validation in Pipelines

Pipelines rarely fail loudly when data is wrong. A job that loads half the usual rows, or a column that quietly fills with nulls, finishes green, and the error surfaces weeks later in a report. Data quality testing makes those failures loud: explicit assertions about the data, run on every load, that stop bad data before it is published.

What to test: six dimensions

• Completeness: required values are present.
• Validity: values are of the right type, in the allowed set or range.
• Uniqueness: no duplicate keys or rows.
• Consistency: values agree with each other, such as an end time after a start time.
• Integrity: every foreign key has a matching parent row.
• Timeliness and volume: the data covers the expected period, and there is roughly as much of it as usual.

A test, in dbt's formulation, is an assertion written as a SQL query that selects failing records; if it returns zero rows, it passes (dbt docs: data tests). That shape, count the rows that break the rule, works in any engine.

Nine checks on real data

The script applies that pattern to January 2024 New York taxi trips (NYC TLC: trip record data). Allowed codes come from the TLC's data dictionary (TLC: data dictionary); each check has a tolerance, the share of rows allowed to fail.

import urllib.request
from pathlib import Path
import duckdb

base = "https://d37ci6vzurychx.cloudfront.net"
for name, url in {"yellow_tripdata_2024-01.parquet": f"{base}/trip-data/yellow_tripdata_2024-01.parquet",
                  "taxi_zone_lookup.csv": f"{base}/misc/taxi_zone_lookup.csv"}.items():
    if not Path(name).exists():
        urllib.request.urlretrieve(url, name)
con = duckdb.connect()
con.execute("CREATE TABLE trips AS SELECT * FROM 'yellow_tripdata_2024-01.parquet'")
con.execute("CREATE TABLE zones AS SELECT * FROM read_csv('taxi_zone_lookup.csv', header = true)")
total = con.sql("SELECT COUNT(*) FROM trips").fetchone()[0]

# each check: (dimension, name, SQL counting the rows that fail, share of rows allowed to fail)
checks = [
    ("completeness", "passenger_count is not null", "SELECT COUNT(*) FROM trips WHERE passenger_count IS NULL", 0.0),
    ("validity", "payment_type is a known code", "SELECT COUNT(*) FROM trips WHERE payment_type NOT IN (0, 1, 2, 3, 4, 5, 6)", 0.0),
    ("validity", "RatecodeID is a known code", "SELECT COUNT(*) FROM trips WHERE RatecodeID NOT IN (1, 2, 3, 4, 5, 6, 99)", 0.0),
    ("validity", "fare_amount is not negative", "SELECT COUNT(*) FROM trips WHERE fare_amount < 0", 0.02),
    ("consistency", "dropoff is not before pickup", "SELECT COUNT(*) FROM trips WHERE tpep_dropoff_datetime < tpep_pickup_datetime", 0.001),
    ("timeliness", "pickup falls in January 2024", "SELECT COUNT(*) FROM trips WHERE tpep_pickup_datetime < '2024-01-01' OR tpep_pickup_datetime >= '2024-02-01'", 0.001),
    ("uniqueness", "no duplicate rows", "SELECT COUNT(*) - COUNT(DISTINCT trips) FROM trips", 0.0),
    ("integrity", "pickup zone exists in lookup", "SELECT COUNT(*) FROM trips t ANTI JOIN zones z ON z.LocationID = t.PULocationID", 0.0),
]
print(f"{total} rows checked")
print(f"{'dimension':<13}{'check':<31}{'failing':>8}{'share':>8}  result")
failed = 0
for dimension, name, sql, allowed in checks:
    failing = con.sql(sql).fetchone()[0]
    ok = failing / total <= allowed
    failed += not ok
    print(f"{dimension:<13}{name:<31}{failing:>8}{failing / total:>8.2%}  {'pass' if ok else 'FAIL'}")

# volume: every day should be present and none far below the typical day
days = con.sql("SELECT CAST(tpep_pickup_datetime AS DATE) AS d, COUNT(*) AS n FROM trips WHERE tpep_pickup_datetime >= '2024-01-01' AND tpep_pickup_datetime < '2024-02-01' GROUP BY 1").fetchall()
counts = sorted(n for _, n in days)
median = counts[len(counts) // 2]
low = sorted((str(d), n) for d, n in days if n < 0.7 * median)
print(f"volume: {len(days)} of 31 days present, median {median} trips a day, days below 70% of median: {low}")
print(f"{failed} of {len(checks)} checks failed")

2964624 rows checked
dimension    check                           failing   share  result
completeness passenger_count is not null      140162   4.73%  FAIL
validity     payment_type is a known code          0   0.00%  pass
validity     RatecodeID is a known code            0   0.00%  pass
validity     fare_amount is not negative       37448   1.26%  pass
consistency  dropoff is not before pickup         56   0.00%  pass
timeliness   pickup falls in January 2024         18   0.00%  pass
uniqueness   no duplicate rows                     0   0.00%  pass
integrity    pickup zone exists in lookup          0   0.00%  pass
volume: 31 of 31 days present, median 97117 trips a day, days below 70% of median: [('2024-01-07', 67543)]
1 of 8 checks failed

Reading the results

One check fails, and it is informative. 140,162 trips, 4.73%, have no passenger count. A threshold of zero says that is unacceptable; whether it truly is depends on what the column is used for. The useful next step is to ask why: here every one of those rows is a trip recorded with payment type 0, and no other trips are affected, so the nulls belong to one category of record, not random corruption. Checks do not only block bad data; they teach you the data.

Tolerances turn noise into signal. Negative fares are 1.26% of rows. With a tolerance of zero this check would fail every month and be ignored; with a tolerance of 2% it passes today and fails the day the share jumps, which is the event worth an alert. The same applies to the 56 trips that end before they start and the 18 dated outside the month.

Valid codes, unique rows and referential integrity pass, which is worth recording too: these are the assertions that protect joins and counts downstream.

The volume check finds something row checks cannot. Every day of the month is present, but 7 January has 67,543 trips against a median of 97,117. No single row is wrong; the day as a whole is unusual. It may be genuine, or a sign that part of a day's data is missing. Either way, a person should look before the number reaches a dashboard.

One subtlety in the code: NOT IN never matches null values, so the rate code check does not flag rows where the code is missing. Pair accepted-value checks with not-null checks, or the nulls slip between them.

Thresholds and severity

• Error: stop the pipeline. Use for broken keys, schema changes, empty loads and anything that would make published numbers wrong.
• Warn: publish, but alert. Use for drifting rates and unusual volumes.
• Base thresholds on history, not guesses: record each check's failing share on every run, and alert on departures from the usual range.

dbt exposes this as severity with warn and error thresholds, and can store failing rows in a table for inspection.

Where checks belong

• At ingestion: schema, row count and freshness, before anything is transformed.
• Between layers: validity and completeness before data enters the cleaned layer; medallion architecture shows a quarantine for rows that fail.
• Before publishing: reconciliation of totals against the source, and business rules.
• In code review: tests run in continuous integration on sample data, so a transformation change that breaks an assertion never reaches production.

Building your first data pipeline shows the simplest version, rules counted on every run.

Tools

The checks above are plain SQL in a loop, which is enough to start. Dedicated tools add packaging, history and reporting:

• dbt tests ship four generic tests, unique, not_null, accepted_values and relationships, declared in configuration beside each model; the dbt tutorial covers them.
• Great Expectations expresses checks as Expectations, verifiable assertions about data, grouped into suites, run through checkpoints and rendered as readable documentation (GX docs: overview).
• Platform features such as constraints and expectations in warehouses and lakehouse pipelines enforce rules at write time.

Agreeing the rules with the teams that produce the data is the subject of data contracts.

Data quality and validation are part of the data governance and security module, and pipeline monitoring and logging of the monitoring module, in the Data Engineering course.

Sources

• Add data tests to your DAG (dbt documentation) — dbt Labs
• GX Core overview (Great Expectations documentation) — Great Expectations
• Data dictionary: yellow taxi trip records (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission