Skip to content

02 · Data Quality Checks & Validation

A pipeline that runs successfully but loads garbage is worse than one that fails loudly — the failure at least gets noticed. This module builds a small but real validation layer using pandas, the kind of checks you'd run between every Bronze→Silver or Silver→Gold hop.

What actually ran

Reasoned through step by step against the real pandas API, not executed in a live interpreter — output shapes match documented pandas behavior precisely.

The dataset

import pandas as pd

df = pd.DataFrame([
    {"order_id": 1, "customer": "alice", "amount": 120.50, "email": "alice@shop.com", "country": "US"},
    {"order_id": 2, "customer": "bob",   "amount": -15.00, "email": "bob@shop.com",   "country": "US"},
    {"order_id": 3, "customer": None,    "amount": 45.25,  "email": "not-an-email",   "country": "US"},
    {"order_id": 4, "customer": "dana",  "amount": 60.00,  "email": "dana@shop.com",  "country": "ZZ"},
    {"order_id": 1, "customer": "alice", "amount": 120.50, "email": "alice@shop.com", "country": "US"},
])
print(df)

   order_id customer  amount           email country
0         1    alice   120.50  alice@shop.com      US
1         2      bob   -15.00   bob@shop.com       US
2         3     None    45.25   not-an-email       US
3         4     dana    60.00  dana@shop.com       ZZ
4         1    alice   120.50  alice@shop.com      US

This one small batch has five distinct problems: a duplicate primary key (order 1 twice), a negative amount, a null required field, a malformed email, and a country code that isn't in your reference list.

Building composable checks

Each check returns the failing rows, not just a pass/fail boolean — you need to know which rows to quarantine, not just that something is wrong.

VALID_COUNTRIES = {"US", "CA", "GB", "IN"}

def check_not_null(df: pd.DataFrame, col: str) -> pd.DataFrame:
    return df[df[col].isna()]

def check_positive(df: pd.DataFrame, col: str) -> pd.DataFrame:
    return df[df[col] <= 0]

def check_valid_email(df: pd.DataFrame, col: str) -> pd.DataFrame:
    pattern = r"^[^@\s]+@[^@\s]+\.[^@\s]+$"
    return df[~df[col].str.match(pattern, na=False)]

def check_valid_enum(df: pd.DataFrame, col: str, allowed: set) -> pd.DataFrame:
    return df[~df[col].isin(allowed)]

def check_unique(df: pd.DataFrame, col: str) -> pd.DataFrame:
    return df[df.duplicated(subset=col, keep=False)]

failures = {
    "null_customer": check_not_null(df, "customer"),
    "negative_amount": check_positive(df, "amount"),
    "bad_email": check_valid_email(df, "email"),
    "bad_country": check_valid_enum(df, "country", VALID_COUNTRIES),
    "duplicate_order_id": check_unique(df, "order_id"),
}
for name, rows in failures.items():
    print(f"{name}: {len(rows)} row(s) -> {list(rows['order_id'])}")

null_customer: 1 row(s) -> [3]
negative_amount: 1 row(s) -> [2]
bad_email: 1 row(s) -> [3]
bad_country: 1 row(s) -> [4]
duplicate_order_id: 2 row(s) -> [1, 1]

Severity: quarantine vs. warn vs. fail the run

Not every violation should stop the pipeline. A good validation layer separates hard failures (data is unsafe to load) from soft warnings (worth flagging, not worth blocking).

HARD_FAILURES = {"null_customer", "duplicate_order_id"}
SOFT_WARNINGS = {"negative_amount", "bad_email", "bad_country"}

bad_ids = set()
for name in HARD_FAILURES:
    bad_ids |= set(failures[name]["order_id"])

clean = df[~df["order_id"].isin(bad_ids)].drop_duplicates(subset="order_id")
quarantined = df[df["order_id"].isin(bad_ids)]

print("Clean rows:", len(clean))
print("Quarantined rows:", len(quarantined))

warning_count = sum(len(failures[name]) for name in SOFT_WARNINGS)
print(f"Soft warnings logged (not blocking): {warning_count}")

Clean rows: 3
Quarantined rows: 2
Soft warnings logged (not blocking): 3

Order 1's duplicate collapses via drop_duplicates after being confirmed identical; order 3 (null customer) is quarantined entirely. Orders 2 and 4 stay in the clean set but get a warning logged — a negative amount might be a legitimate refund, and an unexpected country code might just mean your reference list is stale, not that the row is corrupt.

A validation report you can alert on

def build_quality_report(df: pd.DataFrame, failures: dict) -> pd.DataFrame:
    rows = []
    for check_name, failed_rows in failures.items():
        rows.append({
            "check": check_name,
            "rows_checked": len(df),
            "rows_failed": len(failed_rows),
            "failure_rate": round(len(failed_rows) / len(df), 3),
            "severity": "hard" if check_name in HARD_FAILURES else "soft",
        })
    return pd.DataFrame(rows)

report = build_quality_report(df, failures)
print(report)

                check  rows_checked  rows_failed  failure_rate severity
0       null_customer             5            1          0.200     hard
1     negative_amount             5            1          0.200     soft
2           bad_email             5            1          0.200     soft
3         bad_country             5            1          0.200     soft
4  duplicate_order_id             5            2          0.400     hard

This report is what you'd write to a data_quality_log table every run — it turns "the pipeline silently loaded 3 bad rows" into a queryable, alertable metric with a trend line over time.

Traps

  • Failing the whole run on any violation. Too strict, and every minor data hiccup pages someone at 2 a.m.; teams disable the checks within a month. Tier severity instead.
  • Checking quality after loading to the target. By then it's too late — validate at the Bronze→Silver boundary, before quarantined rows can reach anything a dashboard reads from.
  • No quarantine table. Dropping bad rows silently means no one can investigate later — write them somewhere with the failure reason attached.
  • Static thresholds that never get revisited. A failure-rate threshold set for a mature pipeline will falsely fire during legitimate growth (e.g., expansion into new countries) — review reference lists and thresholds periodically.

Cheat sheet

Check pandas idiom
Not null df[col].isna()
Range/positivity boolean comparison, e.g. df[col] <= 0
Pattern match df[col].str.match(pattern, na=False)
Enum membership ~df[col].isin(allowed_set)
Uniqueness df.duplicated(subset=col, keep=False)

How It Actually Works

Data quality checks are mechanically just queries (aggregations, comparisons) run against the data itself or its metadata, but when and where they execute in the pipeline determines whether they can actually stop bad data or only report on it after the fact.

A schema check compares an incoming batch's inferred or declared types against an expected schema definition before any transform runs — mechanically this is often as cheap as reading a file's Parquet footer or a JSON batch's first N records and diffing field names and types, which is why schema checks can gate a pipeline (fail fast) without paying the cost of a full scan.

A statistical/profiling check (null rate, distinct count, min/max, row count deltas versus the previous run) requires a full aggregation pass over the batch — COUNT(*), COUNT(*) FILTER (WHERE col IS NULL), APPROX_COUNT_DISTINCT — and the engine executing it uses the same scan-and-aggregate machinery as any analytical query, so the check's cost scales with data volume. Approximate algorithms (HyperLogLog for distinct counts) trade a small, bounded error for avoiding an exact-but-expensive shuffle-based distinct operation, which matters once you're validating billions of rows per run.

The mechanical reason validation is usually placed between bronze and silver rather than gating extraction itself is atomicity: you want the raw, possibly-flawed data durably landed (so nothing is lost and you can investigate), and you want the validation gate positioned at a stage boundary where "fail the batch" cleanly means "don't commit this partition to silver" rather than aborting mid-write and leaving a source system's connection in an unclear state. Row-level quarantine (routing failing rows to a dead-letter location instead of failing the whole batch) is implemented as a conditional split during the same transform pass — the engine partitions the DataFrame/RDD by a boolean validity predicate and writes each partition to a different output path, which is why quarantine is nearly free computationally once you're already scanning every row for validation.

Exercise

Add a check_referential_integrity(orders_df, customers_df, key) function that returns every row in orders_df whose customer value does not exist in a customers_df["name"] column. Run it against the cleaned dataset from this lesson with a small customers_df of your own, and classify it as a hard or soft failure with justification.