Data Contracts for Pipeline Teams: Fail the Producer, Not Just the Test

A dbt not_null test on fct_orders is useful. It is also late. The producer already wrote the file, the topic, or the landing table. The consumer DAG failed at 07:10. Finance saw a hole. The warehouse test did its job and still did not prevent the mess.

A data contract is the agreement the producer is willing to fail their own job for: schema, grain, freshness, and a volume band. Consumer tests stay. They are not the contract.

The check primitives (nulls, uniqueness, ranges, freshness) are in Data quality checks in Python. This post is where those checks live — on the write path — and who gets paged.

flowchart LR
  prod[Producer job] --> gate[Contract gate]
  gate -->|pass| land[Landing / topic / table]
  gate -->|fail| page[Page producer]
  land --> cons[Consumer / dbt]
  cons --> tests[Warehouse tests]
  tests -->|fail| late[Late alarm]

The contract fails the write. Warehouse tests catch other writers and drift. They should not be the first alarm.

Schema plus SLAs is the contract

A schema file without time is a type hint. Pipelines break on when and how much as often as they break on types.

A contract that is worth enforcing names:

1. Grain — the business key and uniqueness.
2. Schema — required columns, types, allowed nulls, enums.
3. Freshness — max lag from event time or from the promised arrival clock.
4. Volume — a band, not an exact count.
5. Owner — a rotation that can stop the publish.

# contracts/orders_v1.yaml
name: orders
version: 1
owner: orders-platform
produces:
  uri: s3://lake/landing/orders/dt={ds}/
  format: parquet
schema:
  grain: [order_id]
  columns:
    - { name: order_id, type: string, required: true }
    - { name: customer_id, type: string, required: true }
    - { name: status, type: string, required: true, enum: [placed, paid, cancelled] }
    - { name: amount, type: float64, required: true, min: 0 }
    - { name: ordered_at, type: timestamp, required: true }
    - { name: updated_at, type: timestamp, required: true }
sla:
  arrival_by: "06:00"
  timezone: UTC
  max_lag_hours: 26
  min_rows: 1000
  max_rows: 5000000
compat:
  additive_columns: allow
  rename: breaking
  type_narrow: breaking

arrival_by is a clock SLA (the drop must exist by 06:00 UTC). max_lag_hours is a data SLA (the newest updated_at cannot be stale). You often need both. A file that arrives on time with Tuesday’s leftovers fails freshness. A file that never arrives fails the clock.

Load this YAML in the producer. Changing min_rows in chat is how the contract becomes decoration — same rule as the quality-check config in the Python post.

Fail the producer

The producer job validates before it publishes the partition or commits the topic offsets. Failure means: no _SUCCESS, no new consumer pointer, no “soft” write into gold.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pandas as pd
import yaml


class ContractBreach(Exception):
    pass


@dataclass(frozen=True)
class OrdersContract:
    grain: tuple[str, ...]
    required: tuple[str, ...]
    enums: dict[str, frozenset[str]]
    min_rows: int
    max_rows: int
    max_lag: timedelta
    arrival_by_hour: int


def load_contract(path: Path) -> OrdersContract:
    raw = yaml.safe_load(path.read_text(encoding="utf-8"))
    sla = raw["sla"]
    schema = raw["schema"]
    enums = {
        c["name"]: frozenset(c["enum"])
        for c in schema["columns"]
        if "enum" in c
    }
    return OrdersContract(
        grain=tuple(schema["grain"]),
        required=tuple(c["name"] for c in schema["columns"] if c.get("required")),
        enums=enums,
        min_rows=int(sla["min_rows"]),
        max_rows=int(sla["max_rows"]),
        max_lag=timedelta(hours=int(sla["max_lag_hours"])),
        arrival_by_hour=int(sla["arrival_by"].split(":")[0]),
    )


def assert_contract(df: pd.DataFrame, cfg: OrdersContract, now: datetime) -> None:
    failures: list[str] = []
    n = len(df)
    if n < cfg.min_rows:
        failures.append(f"volume {n} < {cfg.min_rows}")
    if n > cfg.max_rows:
        failures.append(f"volume {n} > {cfg.max_rows}")

    missing = [c for c in cfg.required if c not in df.columns]
    if missing:
        raise ContractBreach(f"missing columns: {missing}")

    for col in cfg.grain:
        if df[col].duplicated().any() or df[col].isna().any():
            failures.append(f"grain broken on {col}")

    for col, allowed in cfg.enums.items():
        bad = set(df[col].dropna().astype(str)) - allowed
        if bad:
            failures.append(f"{col} has values {sorted(bad)}")

    newest = pd.to_datetime(df["updated_at"], utc=True).max()
    if newest < now - cfg.max_lag:
        failures.append(f"freshness {newest.isoformat()}")

    if failures:
        raise ContractBreach("; ".join(failures))


def publish_if_ok(df: pd.DataFrame, cfg: OrdersContract, dest: Path) -> None:
    assert_contract(df, cfg, datetime.now(timezone.utc))
    dest.parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(dest, index=False)
    (dest.parent / "_SUCCESS").write_text("ok\n", encoding="utf-8")

No _SUCCESS means the Airflow sensor downstream does not start. That is the point. The consumer should not be the first process that knows the drop is unusable.

Do not quarantine volume or freshness by dropping rows. Those are batch-level. You cannot reject “Monday never arrived” one row at a time. Quarantine is for messy-but-useful rows after the batch itself is a valid publish — see the quality-check post.

Warehouse tests are the second line

Keep dbt tests. Other jobs write. Humans run INSERT. Incremental models drift. The contract does not make the warehouse omniscient.

What changes is who fails first and who is on the page:

Breach	First fail	Page
Missing column, broken grain, empty weekday file	Producer publish	Producer owner
File never arrived by SLA	Producer SLA probe / sensor	Producer owner
Second writer broke uniqueness	dbt test on gold	Both, then fix writers
Consumer model joined wrong	Consumer CI	Consumer

If only the consumer DAG pages, the producer learns about a rename from a Slack thread titled “fct_orders red.” That is not a contract. That is a courtesy.

-- consumer-side companion — still required
select order_id, count(*) as n
from analytics.marts.fct_orders
group by 1
having count(*) > 1
limit 20;

If that returns rows, a producer contract on today’s slice can still be green. Pair the layers. Incremental slices are unique while the target is not — leftover keys from last month.

dbt tests and source freshness belong next to models; the dbt tutorial is the consumer half. Put a freshness block on the source you do not own. Put the publish gate on the job you do own.

Versioning: additive vs breaking

version: 1 is not a comment. Consumers pin it. Breaking changes ship as v2 (new path, new topic, new schema registry id), with a dual-write window if you cannot stop the world.

Additive: a new optional column. Old readers ignore it. New readers use it. Still document it.

Breaking: rename, type change, grain change, required field added without a default, enum value removed, partition clock changed.

def assert_compatible(old: dict, new: dict) -> None:
    if new["version"] < old["version"]:
        raise ContractBreach("version moved backwards")
    old_cols = {c["name"]: c for c in old["schema"]["columns"]}
    new_cols = {c["name"]: c for c in new["schema"]["columns"]}
    dropped = set(old_cols) - set(new_cols)
    if dropped:
        raise ContractBreach(f"removed columns {sorted(dropped)} require a new version")
    if old["schema"]["grain"] != new["schema"]["grain"]:
        raise ContractBreach("grain change is breaking")
    for name, col in old_cols.items():
        nxt = new_cols[name]
        if nxt["type"] != col["type"]:
            raise ContractBreach(f"{name} type {col['type']} -> {nxt['type']}")

Run that in the producer CI when the YAML changes. A PR that renames amount to amount_usd without bumping the version fails CI — not the next morning’s mart.

Do not “fix” a breaking change by making the consumer job defensive (if 'amount' in df.columns). That trains the producer to skip the version bump.

SLA probes are jobs, not vibes

A clock SLA needs a job that runs at arrival_by + a grace minute and checks that the partition exists and passed the contract. Absence is a producer failure.

def sla_probe(success_path: Path, cfg: OrdersContract, now: datetime) -> None:
    deadline = now.replace(
        hour=cfg.arrival_by_hour, minute=0, second=0, microsecond=0
    )
    if now >= deadline and not success_path.exists():
        raise ContractBreach(f"no publish at {success_path} after {deadline.isoformat()}")

Wire the probe in the same repo as the producer. The consumer sensor waiting on _SUCCESS is a dependency edge, not the SLA definition. If you only have the sensor, you have a timeout, not a contract.

flowchart TD
  extract[Producer extract] --> validate[assert_contract]
  validate -->|ok| write[Write partition + _SUCCESS]
  validate -->|breach| pageP[Page producer]
  write --> probe[SLA probe at arrival_by]
  probe -->|missing| pageP
  write --> consumer[Consumer DAG]
  consumer --> dbt[dbt tests]
  dbt -->|other writers / drift| pageC[Page consumer + producer]

Publish, then probe the clock. Consumers start from _SUCCESS, not from hope.

Dual-write when you must bump

v2 is a new URI or topic, not a silent overwrite of v1. For a week (or a month — size it from consumer deploy time, not from optimism) the producer writes both paths after both contracts pass. Consumers cut over. Then you stop v1.

Do not dual-write “until someone remembers.” Put a calendar on the v1 path and a probe that still pages if v1 breaks during the window. Consumers still on v1 are why the old contract exists.

Schema Registry / Avro / Protobuf are the same idea with a central id: compatible evolution is additive; a grain change is a new subject. YAML in the producer repo is enough until you have more than a handful of publishers. Do not wait for a registry to fail the write.

When a contract is the wrong tool

• You do not know the schema yet (a vendor dump you are still profiling). Freeze rules after a week of observation, not on day one.
• There is no owner. A YAML file without a rotation is a wiki page.
• The pipeline is a one-off laptop pull. len(df) is enough.
• You are trying to enforce business definitions (what “revenue” means) across six marts. That is semantic modeling, not a landing-file contract. Keep the contract at the publish edge.

Pitfalls

• Tests only in the warehouse. The producer never fails. You discover the rename in BI.
• Exact row counts. They flap. Use a band. Holiday Tuesdays are not incidents.
• Contract in a Confluence page. The job cannot raise ContractBreach on a paragraph.
• Consumer owns the YAML. Then a breach becomes a consumer ticket. Put the file in the producer repo.
• Quarantining a missing file. There is nothing to quarantine. Fail the publish.
• v1 forever while columns rename. Compatibility CI exists so you bump to v2 on purpose.

Production checklist

• Contract YAML lives in the producer repo and is loaded by the publish job.
• Grain, required columns, volume band, freshness, and arrival_by are explicit.
• Publish writes _SUCCESS (or commits offsets) only after assert_contract.
• SLA probe pages the producer if the drop is late or absent.
• Consumer DAGs wait on the success marker; they do not guess.
• dbt / warehouse tests remain for other writers and for gold grain.
• Breaking changes bump a version and a path. CI rejects silent incompatibilities.
• Alerts name the owner in the contract, not “data-platform” as a junk drawer.

FAQ

Why not only dbt tests? They run after the publish. The bad file is already in the lake and the consumer may have started. Fail the producer first. Keep dbt as the second line.

Who owns the contract file? The team that writes the partition or topic. Consumers may propose changes. They do not silently loosen min_rows to make their DAG green.

Fail the job or quarantine the row? Fail the job on grain, schema, volume, freshness, and a missed SLA. Quarantine messy-but-useful rows only when the batch itself is a valid publish. Do not quarantine “Monday never arrived.”

When is a new column a new version? Optional additive columns can stay on v1 if old readers ignore them. Renames, type changes, grain changes, and new required fields without defaults are v2.

Where do checks run if we already use Great Expectations? Same place: on the producer write path, before _SUCCESS. The catalog of rules can be GX, Soda, or the YAML above. The failure must stop the publish, not only decorate a report.

Quality checks tell you what to assert. A contract tells you when (before publish) and who (the producer). Put the gate on the write, page the owner named in the YAML, and leave warehouse tests in place for everyone else who can still write the table. That is the whole design. The rest is file format.