Picture the 2 AM PagerDuty alert every data engineer dreads: the revenue dashboard showing zeros, the CFO already emailing the CTO, fourteen people in the incident channel, and "the data team broke something again" being said out loud. The root cause is often trivial and entirely preventable - an upstream microservice team renames a field from transaction_amount to txn_amount in an event payload, with no announcement, no migration period, no versioning. One Friday deploy nukes three downstream pipelines, two ML feature stores, and executive reporting, and the blast radius takes days to remediate. Data contracts exist to make that class of incident impossible.
What a data contract is
A data contract is a formal agreement between a data producer and its consumers defining the structure, semantics, quality guarantees, and SLAs of a data interface - an API contract, but for data pipelines instead of REST endpoints. Web development settled this years ago: nobody ships a REST API without documented schemas or renames a public field without a deprecation period. The data world has largely lived in the Wild West, where a Kafka topic's schema is whatever the last developer pushed. A complete contract specifies the schema (field names, types, nullability, nesting), the semantic meaning of each field (is amount in cents or dollars? is timestamp UTC or local?), quality expectations (null rates, value ranges, uniqueness, freshness), SLAs (arrival timing, update frequency, latency), ownership (who produces it, who to call), and versioning rules for how it can evolve without breaking consumers.
Without these written down and enforced, every pipeline is one upstream change from an incident. The failure modes recur: silent schema drift (a type changes from int to string and downstream jobs produce garbage instead of failing); semantic ambiguity (two teams read created_at differently and build correct-looking dashboards that disagree); undocumented deprecation (a field stops being populated but stays in the schema, so consumers query nulls); cascading failures across an unmapped DAG of downstream tables; and a finger-pointing culture where every incident is "the data team should have validated" versus "the app team should have announced." This matters more than it used to: a decade ago one warehouse team owned everything end to end and the contract lived in their heads, but modern stacks have dozens of producers feeding consumers who have never met each other, and the implicit contract doesn't scale.
Approach 1: Pydantic models as contracts
The simplest start is defining expected schemas as code. In Python, Pydantic gives runtime validation, documentation, and serialization for free:
from pydantic import BaseModel, Field, field_validator
from datetime import datetime
from enum import Enum
from uuid import UUID
class CurrencyCode(str, Enum):
USD = "USD"; EUR = "EUR"; GBP = "GBP"
class TransactionEvent(BaseModel):
"""
Contract for payments.transactions.v2 Kafka topic.
Owner: Payments Team (#payments-eng). SLA: events within 30s of completion.
"""
event_id: UUID = Field(description="Unique per event, used for deduplication")
transaction_id: str = Field(min_length=10, max_length=64)
user_id: int = Field(gt=0)
amount_cents: int = Field(ge=0, description="Amount in cents, NOT dollars")
currency: CurrencyCode
status: str = Field(pattern=r"^(completed|refunded|failed|pending)$")
merchant_name: str = Field(min_length=1, max_length=255)
occurred_at: datetime = Field(description="UTC timestamp of the transaction")
processed_at: datetime = Field(description="UTC timestamp we processed it")
@field_validator("occurred_at", "processed_at")
@classmethod
def must_be_utc(cls, v: datetime) -> datetime:
if v.tzinfo is None:
raise ValueError("Timestamps must be timezone-aware (UTC)")
return vThis isn't just a schema. The docstring documents ownership and SLAs, the validators encode business rules beyond type checking, and the field descriptions kill semantic ambiguity (amount is in cents). Because it's code, it lives in version control and goes through review. Validate every incoming event against it - but make violations loud, not blocking:
def process_message(raw: bytes):
try:
return TransactionEvent(**json.loads(raw))
except ValidationError as e:
logger.error("contract_violation", topic="payments.transactions.v2", errors=e.errors())
publish_to_dlq("payments.transactions.v2.dlq", raw, str(e)) # recoverable
CONTRACT_VIOLATIONS.labels(topic="payments.transactions.v2").inc() # alert on rate
return NoneA single malformed event goes to a dead letter queue and increments a counter; the pipeline keeps running. When the violation rate crosses a threshold (1% over 5 minutes), PagerDuty fires - so one bad record doesn't stop the world, but a systematic break is caught fast.
Approach 2: schema registry with Protobuf
Pydantic works within one team's codebase but doesn't solve the cross-team problem: separate repos and deploy cycles need a shared source of truth. A schema registry (Protobuf with Confluent Schema Registry, say) provides it, and adds the thing Pydantic can't - compatibility checking at registration time:
// proto/payments/v2/transaction_event.proto
syntax = "proto3";
package payments.v2;
import "google/protobuf/timestamp.proto";
message TransactionEvent {
string event_id = 1; // UUID, unique per event
int64 user_id = 3; // always > 0
int64 amount_cents = 4; // cents, >= 0
CurrencyCode currency = 5;
TransactionStatus status = 6;
google.protobuf.Timestamp occurred_at = 9; // UTC
}
enum CurrencyCode { CURRENCY_UNSPECIFIED = 0; USD = 1; EUR = 2; GBP = 3; }The registry can enforce BACKWARD (new schema reads old data - consumers upgrade first), FORWARD (old schema reads new data - producers upgrade first), or FULL (both, the safest). With FULL compatibility, an attempt to rename transaction_amount to txn_amount is rejected at registration - the broken schema never reaches Kafka, and you catch it in the PR instead of at 2 AM.
Approach 3: Great Expectations for quality contracts
Schema contracts cover names and types; they don't cover content. A perfectly typed payload can still be garbage. Great Expectations adds a quality layer that runs as a step in the Airflow DAG, right after ingestion and before downstream transformations:
suite = context.suites.add(gx.ExpectationSuite(name="transactions_quality_contract"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="transaction_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(
column="merchant_category_code", mostly=0.85)) # tolerated null rate
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
column="amount_cents", min_value=0, max_value=100_000_00))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="event_id"))
suite.add_expectation(gx.expectations.ExpectColumnDistinctValuesToBeInSet(
column="currency", value_set=["USD", "EUR", "GBP"]))
suite.add_expectation(gx.expectations.ExpectColumnMaxToBeBetween(
column="occurred_at", min_value="now - 48h", max_value="now + 1h")) # freshness
suite.add_expectation(gx.expectations.ExpectTableRowCountToBeBetween(
min_value=10_000, max_value=500_000)) # volume anomalyIf the quality contract fails, the DAG halts, the bad partition is quarantined, and on-call gets a report of exactly which expectations failed. The suite doubles as documentation - a new engineer can read it and immediately know what "good data" looks like, with no guessing whether nulls in merchant_category_code are normal.
Which approach to use
| Aspect | Pydantic (code) | Schema registry | Great Expectations |
|---|---|---|---|
| Validates | Schema + basic rules | Schema + compatibility evolution | Quality + statistical properties |
| Catches issues at | Runtime (per record) | Deploy (schema registration) | Batch (post-ingestion) |
| Cross-team enforcement | Weak (in consumer code) | Strong (shared registry) | Medium (shared suites) |
| Setup complexity | Low | High (registry infra) | Medium |
| Best for | Single-team pipelines | Multi-team streaming | Warehouse/lake quality gates |
You probably need at least two. A common combination is Protobuf + registry at the Kafka boundary, Pydantic in Python consumers for runtime validation, and Great Expectations in the warehouse for quality gates - each layer catching a different class of problem.
Enforce it in CI/CD
Contracts that aren't enforced automatically are just documentation, and documentation gets ignored. The highest-impact move is validating schema compatibility on every PR that touches a contract - checking proto compatibility against the registry, running the Pydantic contract tests, and posting a human-readable contract-diff comment so reviewers who don't read Protobuf fluently still understand the impact:
# .github/workflows/schema-check.yml (sketch)
on: { pull_request: { paths: ['proto/**/*.proto', 'contracts/**/*.py'] } }
jobs:
check-compatibility:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- run: python scripts/check_schema_compat.py --compatibility FULL --files ${CHANGED}
- run: pytest tests/contracts/ -v # validate Pydantic contracts
- run: python scripts/contract_diff.py --base origin/main --head HEAD # PR commentThe most valuable contract test encodes the original incident directly - if anyone renames amount_cents, it fails before reaching any environment:
def test_contract_rejects_renamed_field():
payload = {"event_id": "550e8400-e29b-41d4-a716-446655440000",
"transaction_id": "TXN-1234567890", "user_id": 42,
"txn_amount": 1999, # wrong field name!
"currency": "USD", "status": "completed", "merchant_name": "Coffee Shop",
"occurred_at": "2025-12-01T10:30:00Z", "processed_at": "2025-12-01T10:30:05Z"}
with pytest.raises(Exception):
TransactionEvent(**payload)Run contract tests against anonymized production samples, not synthetic data - synthetic data always passes, while real samples catch edge cases you'd never think to generate.
The organizational problem
The hardest part isn't the technology - it's getting people to care. "That sounds like more process," "we move fast, we can't add friction to every deploy," "the data team should just handle schema changes" are the predictable first reactions. Three things turn that around. Quantify the pain: tally the data incidents from the last year (one audit found 47 incidents, 8.3 hours average to detect, 14.6 to resolve - roughly $340,000 in engineering time, before counting wrong-dashboard business impact), and the conversation shifts from "why" to "how fast." Start with the highest-pain interface: don't boil the ocean - put a contract on the single worst topic, catch a few would-be breaking changes in review, and let other teams ask for their own. Make it easy, not bureaucratic: a four-page spec per interface gets ignored; a single Pydantic or Protobuf file with an automatic CI check takes ~15 minutes to create and zero to maintain. One thing worth doing from day one: have producers propose the contract and consumers review it, rather than the consumer side handing producers a fait accompli - the collaborative model gets far better adoption.
A contract spec format
Beyond the schema file itself, a YAML metadata layer captures everything else and becomes the single source of truth during an incident - who owns it, the quality expectations, who to notify, and the process for breaking changes:
# contracts/payments/transactions-v2.yaml
apiVersion: datacontract/v1
kind: DataContract
metadata:
name: payments-transactions-v2
version: "2.4.1"
owner: { team: payments-eng, slack: "#payments-eng" }
schema: { type: protobuf, path: proto/payments/v2/transaction_event.proto, compatibility: FULL }
quality:
freshness: { max_delay: 30s }
completeness: { optional_fields: { merchant_category_code: { max_null_rate: 15% } } }
volume: { daily_min: 10000, daily_max: 500000, anomaly_detection: true }
sla: { availability: 99.9%, incident_response: 30m, planned_maintenance_notice: 48h }
consumers:
- { team: data-platform, usage: "Warehouse ingestion, daily aggregates" }
- { team: ml-engineering, usage: "Real-time fraud-model features" }
change_management:
breaking_changes: { notice_period: 14d, approval_required_from: [data-platform, ml-engineering] }
non_breaking_changes: { notice_period: 3d, auto_approved: true }Monitoring and observability
Contracts without monitoring are wishful thinking. The metrics worth tracking: violation rate (records failing validation by error type, alerting at 1% over 5 minutes); schema-drift events (incompatible-schema registration attempts - even blocked ones flag a team that doesn't know about the contract); freshness lag (p50/p95/p99 between event and warehouse landing); volume anomalies (a rolling 7-day mean with 3-sigma bounds); and contract coverage (share of interfaces with a formal contract - the goal is 100% of tier-1 interfaces). A Grafana table listing every contract as green / yellow / red gives leadership a single pane of glass for data reliability.
Lessons after a year
- Start with schema contracts, add quality later. Agreeing field names and types solves about 70% of incidents; null rates and volume bounds are phase two.
- Version everything. Every contract gets a version; every change increments it. Forgetting to version Pydantic models leaves you unable to tell which contract a consumer runs.
- Dead letter queues are non-negotiable. Failed records must go somewhere recoverable - silently dropping them once cost two days of compliance-relevant data.
- Make producers own their contracts. Consumers review and request changes, but ownership must be clear - shared ownership is no ownership.
- Contracts are living documents. Schedule quarterly reviews; business logic, new fields, and dead fields drift a contract out of date.
- Don't enforce in dev. Log violations everywhere, but ramp enforcement toward production: log in dev, warn in staging, block in prod.
Getting started this week
Pick the single most problematic interface. Write a Pydantic model (or JSON Schema or Protobuf) describing valid data for it, with field descriptions and basic validators. Add validation to the consumer pipeline - logging violations, not blocking - and ship it. Review the violation logs (you'll almost certainly find quality issues you didn't know about) and fix the obvious ones. Then share the contract with the producing team and agree that future schema changes go through PR review. Five days, one contract, value proven - everything after that is scaling what works. Schema-related incidents dropping by roughly 89%, mean time to detect falling from 8.3 hours to 12 minutes, and nobody woken up by a broken pipeline in months is the kind of result that makes the case on its own. Data contracts aren't glamorous, but they're among the highest-ROI investments in data reliability: start small, automate enforcement, and let the results speak.
Leave a Comment