Schema checks: make your pipeline refuse bad data

Every data pipeline lives downstream of people who don’t know it exists. The app team renames user_id to account_id, an export switches from UTC to local time, a currency column starts arriving in cents. None of these throw errors in your pipeline — pandas will happily compute garbage with full confidence. The dashboards stay green while the numbers go wrong, and you find out from an executive instead of a stack trace.

The cure is old-fashioned: contracts, checked at the border. Every dataset entering your pipeline passes through an explicit schema — column names, types, value ranges, nullability, uniqueness — and the pipeline halts on violation.

Asserts are a fine starting point

You don’t need a framework to start. A validation function per input, wired into the load path, catches the majority of upstream surprises:

def validate_orders(df: pd.DataFrame) -> pd.DataFrame:
    expected = {"order_id", "customer_id", "amount", "currency", "created_at"}
    missing = expected - set(df.columns)
    assert not missing, f"schema change: missing {missing}"

    assert df.order_id.is_unique, "duplicate order_ids"
    assert df.amount.between(0.01, 50_000).all(), \
        "amount out of range — cents/currency bug?"
    assert df.currency.isin({"EUR", "USD", "GBP"}).all(), \
        f"new currency: {set(df.currency.unique()) - {'EUR','USD','GBP'}}"
    assert df.created_at.dt.tz is not None, "timestamps lost their timezone"
    return df

Note what these check: not just structure, but semantics — ranges that encode domain knowledge (“no order is €0 or €2M”), enumerations that catch new categories, timezone awareness. Each assert is a piece of tribal knowledge promoted into an executable sentence.

When to graduate to a framework

Hand-rolled asserts sprawl. Two mature options structure them:

Pandera — schemas as code, close to pandas, great for in-process pipelines:

import pandera as pa

OrdersSchema = pa.DataFrameSchema({
    "order_id": pa.Column(str, unique=True),
    "amount": pa.Column(float, pa.Check.in_range(0.01, 50_000)),
    "currency": pa.Column(str, pa.Check.isin(["EUR", "USD", "GBP"])),
    "created_at": pa.Column(pa.DateTime(tz="UTC")),
})
df = OrdersSchema.validate(df, lazy=True)   # collects ALL failures, not just the first

lazy=True matters operationally: one run tells you everything wrong, not the first thing.

Great Expectations / dbt tests — better when validation belongs to the warehouse layer and multiple teams need to see results. dbt’s not_null/unique/accepted_values tests on staging models give you 80% of the value with trivial effort if you’re already in dbt.

The tool matters less than the placement: validate at every boundary you don’t control — external files, API pulls, other teams’ tables — and once after your own final transformations (to catch your bugs, which exist too).

Design decisions that decide whether it survives

Fail closed for structure, alert for statistics. A missing column should stop the pipeline dead. A distribution shift (null rate doubled, new category at 0.1%) should alert but may not need to halt — encode the difference explicitly, or the team will disable the whole layer after its first noisy week.

Quarantine, don’t drop. When row-level checks fail, write offenders to a rejected_rows table with the reason and a timestamp. Silent dropping is how “we validate our data” becomes “we delete whatever disagrees with our assumptions.”

Keep thresholds honest. Every range encodes an assumption that will age. Review the failures monthly: recurring false alarms mean the world changed and the contract needs renegotiating — which is exactly the conversation with the upstream team that the schema was designed to force.

An afternoon of schema-writing buys you the failure mode every data team dreams of: pipelines that break loudly, immediately, and at the right line — instead of politely serving wrong numbers until someone important notices.