It's 9 AM on a Tuesday, and your dbt models are failing in production. The upstream team that owns the CRM ingestion pipeline silently added a new field, renamed customer_id to cust_id, and changed the data type of revenue from NUMERIC to VARCHAR — all in the same deploy. Your downstream models, your BI dashboards, and your machine learning feature pipeline are now returning garbage or nothing at all. You find out about the breakage not from an alert, but from a Slack message that starts with "Hey, why is the revenue report showing zeros?"
This is the classic producer-consumer trust problem in data engineering. Producers (ingestion teams, application teams, event tracking teams) and consumers (transformation teams, analytics engineers, ML engineers) often operate on different schedules, under different incentive structures, with no formal agreement about what the data between them is supposed to look like. Data contracts are the mechanism that changes this. A data contract is a versioned, machine-readable agreement — covering schema, semantics, SLAs, and ownership — that governs what producers will deliver and what consumers can depend on.
By the end of this lesson, you'll be able to design and enforce data contracts across the full ingestion-to-transformation boundary. We'll work through a realistic e-commerce scenario involving order data flowing from a Kafka-based ingestion layer into a Snowflake transformation layer. You'll write contracts in code, enforce them at multiple pipeline stages, handle breaking and non-breaking schema changes, and build a workflow that lets producer and consumer teams collaborate without stepping on each other.
What you'll learn:
You should be comfortable with:
Before you can implement anything, you need a precise mental model of what a data contract is. It's not just a schema. Think of it as a formal interface definition — analogous to an API contract in software engineering — that covers four distinct concerns.
Schema: The fields, their data types, nullability constraints, and expected value ranges. This is what most teams start with, but it's only a quarter of the story.
Semantics: What the data means. A field named status that contains COMPLETED, CANCELLED, and PENDING is ambiguous without documentation about what those states represent, when they transition, and whether historical records can change (i.e., is this a mutable or append-only dataset?).
Quality: Quantified expectations about completeness, uniqueness, and timeliness. For example: "The order_id field is always unique within a 24-hour window. No more than 0.1% of rows will have a null customer_id. Data arrives within 15 minutes of the transaction timestamp."
Ownership and SLAs: Who owns the dataset, who owns the contract, what the producer's uptime commitment is, and what the notification process looks like when changes are needed.
Here's a concrete contract specification for an orders dataset. We'll use a YAML-based format because it's human-readable, git-diffable, and can be parsed programmatically:
# contracts/orders_v1.yaml
contract:
id: orders
version: "1.0.0"
status: active
producer:
team: platform-engineering
contact: platform-data@company.com
repo: github.com/company/order-service
consumer:
team: analytics-engineering
contact: analytics@company.com
repo: github.com/company/analytics
dataset:
name: raw_orders
description: >
Order events published by the order service whenever an order is
created, updated, or reaches a terminal state. One row per event,
not one row per order — the same order_id can appear multiple times
with different statuses.
mutability: append-only
latency_sla_minutes: 15
availability_sla_percent: 99.5
schema:
fields:
- name: order_id
type: STRING
nullable: false
description: UUID v4 identifying the order. Not globally unique in this table.
tests:
- not_null
- matches_regex: "^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
- name: customer_id
type: STRING
nullable: false
description: Internal customer identifier from the identity service.
tests:
- not_null
- name: status
type: STRING
nullable: false
description: Order lifecycle status.
allowed_values:
- PENDING
- CONFIRMED
- SHIPPED
- DELIVERED
- CANCELLED
- REFUNDED
tests:
- not_null
- accepted_values
- name: revenue_usd
type: NUMERIC(12,2)
nullable: true
description: >
Total revenue in USD at the time of this event. Null for
CANCELLED and REFUNDED statuses where revenue is reversed.
tests:
- not_null_when: "status IN ('CONFIRMED', 'SHIPPED', 'DELIVERED')"
- greater_than_or_equal: 0
- name: event_timestamp
type: TIMESTAMP_TZ
nullable: false
description: UTC timestamp when the order service emitted this event.
tests:
- not_null
- not_in_future
- name: ingested_at
type: TIMESTAMP_TZ
nullable: false
description: UTC timestamp added by the ingestion layer on arrival.
tests:
- not_null
quality:
row_count_minimum_per_hour: 100
uniqueness:
- key: [order_id, status, event_timestamp]
constraint: unique
freshness_warn_after_minutes: 20
freshness_error_after_minutes: 60
changelog:
- version: "1.0.0"
date: "2024-01-15"
author: platform-engineering
changes: Initial contract definition.
This contract is the source of truth. It lives in version control, and changes to it go through a pull request process — we'll define what that looks like in the section on evolution.
The first enforcement point is at the source: when events arrive from the producer and before they land anywhere persistent. This is your cheapest enforcement layer because you catch problems when the data is still in motion, before it pollutes your raw tables.
If your ingestion pipeline uses Kafka with Confluent Schema Registry, you get schema enforcement for free by registering an Avro or JSON Schema for your topic. But many teams use lighter-weight ingestion — Fivetran, Airbyte, custom Python scripts, or Webhook receivers — where you need to add validation yourself.
Here's a Python-based validator that reads the contract YAML, builds a JSON Schema from it, and validates incoming events before they're written to Snowflake's raw layer:
# ingestion/contract_validator.py
import json
import re
from datetime import datetime, timezone
from typing import Any
import jsonschema
import yaml
def load_contract(contract_path: str) -> dict:
with open(contract_path, "r") as f:
return yaml.safe_load(f)
def contract_to_json_schema(contract: dict) -> dict:
"""Convert our YAML contract format to a JSON Schema document."""
type_mapping = {
"STRING": "string",
"INTEGER": "integer",
"FLOAT": "number",
"BOOLEAN": "boolean",
"TIMESTAMP_TZ": "string", # represented as ISO 8601 strings in JSON
}
properties = {}
required_fields = []
for field in contract["schema"]["fields"]:
field_name = field["name"]
field_type = field["type"].split("(")[0] # Strip NUMERIC(12,2) -> NUMERIC
json_type = type_mapping.get(field_type, "string")
field_schema: dict[str, Any] = {"type": json_type, "description": field.get("description", "")}
if not field.get("nullable", True):
required_fields.append(field_name)
if "matches_regex" in (field.get("tests") or []):
# Find the regex from the tests list
for test in field["tests"]:
if isinstance(test, dict) and "matches_regex" in test:
field_schema["pattern"] = test["matches_regex"]
if "allowed_values" in field:
field_schema["enum"] = field["allowed_values"]
properties[field_name] = field_schema
return {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"properties": properties,
"required": required_fields,
"additionalProperties": True, # Allow extra fields — they'll be flagged but not rejected
}
def validate_event(event: dict, contract: dict) -> list[str]:
"""
Validate a single event against the contract.
Returns a list of violation strings. Empty list means valid.
"""
violations = []
json_schema = contract_to_json_schema(contract)
validator = jsonschema.Draft7Validator(json_schema)
errors = list(validator.iter_errors(event))
for error in errors:
field_path = ".".join(str(p) for p in error.absolute_path) or "root"
violations.append(f"SCHEMA_VIOLATION [{field_path}]: {error.message}")
# Custom tests that go beyond what JSON Schema supports
violations.extend(_run_custom_tests(event, contract))
return violations
def _run_custom_tests(event: dict, contract: dict) -> list[str]:
violations = []
for field in contract["schema"]["fields"]:
field_name = field["name"]
value = event.get(field_name)
tests = field.get("tests", []) or []
for test in tests:
if test == "not_in_future" and value:
try:
ts = datetime.fromisoformat(value.replace("Z", "+00:00"))
now = datetime.now(timezone.utc)
if ts > now:
violations.append(
f"QUALITY_VIOLATION [{field_name}]: "
f"Timestamp {value} is in the future."
)
except ValueError:
violations.append(
f"PARSE_ERROR [{field_name}]: Cannot parse as timestamp: {value}"
)
if isinstance(test, dict) and "greater_than_or_equal" in test:
if value is not None and value < test["greater_than_or_equal"]:
violations.append(
f"QUALITY_VIOLATION [{field_name}]: "
f"Value {value} is less than minimum {test['greater_than_or_equal']}."
)
return violations
def validate_batch(events: list[dict], contract_path: str) -> dict:
contract = load_contract(contract_path)
results = {"valid": [], "invalid": [], "violation_summary": {}}
for i, event in enumerate(events):
violations = validate_event(event, contract)
if violations:
results["invalid"].append({"index": i, "event": event, "violations": violations})
for v in violations:
violation_type = v.split("[")[0].strip()
results["violation_summary"][violation_type] = (
results["violation_summary"].get(violation_type, 0) + 1
)
else:
results["valid"].append(event)
total = len(events)
invalid_count = len(results["invalid"])
results["validity_rate"] = round((total - invalid_count) / total * 100, 2) if total > 0 else 0
return results
You'd call this from your ingestion script before writing to the warehouse:
# ingestion/orders_ingestor.py
from contract_validator import validate_batch
import snowflake.connector
import json
def ingest_orders(raw_events: list[dict], dry_run: bool = False):
validation_results = validate_batch(raw_events, "contracts/orders_v1.yaml")
print(f"Validated {len(raw_events)} events: "
f"{len(validation_results['valid'])} valid, "
f"{len(validation_results['invalid'])} invalid "
f"({validation_results['validity_rate']}% pass rate)")
if validation_results["invalid"]:
# Write violations to a dead-letter table for investigation
_write_to_dead_letter(validation_results["invalid"])
violation_rate = 100 - validation_results["validity_rate"]
if violation_rate > 5.0:
raise RuntimeError(
f"Contract violation rate {violation_rate:.1f}% exceeds 5% threshold. "
f"Halting ingestion. Check dead-letter table: raw.orders_dead_letter"
)
if not dry_run and validation_results["valid"]:
_write_to_snowflake(validation_results["valid"])
Key decision point: Whether to reject invalid events entirely, quarantine them to a dead-letter table, or pass them through with a violation flag is a business decision. For financial data, reject and alert. For behavioral event data, quarantine and continue. Define this policy in the contract's
on_violationfield so it's explicit.
Once data is in the raw layer, your transformation team takes over. dbt is where you get your second enforcement point — and where most teams currently have some schema tests, just not systematically tied to a contract.
The goal here is to make your dbt tests derived from the contract, not handwritten independently. This ensures the contract remains the single source of truth and your dbt tests don't drift from what was agreed.
Here's a script that reads your contract YAML and generates a dbt schema.yml file:
# tools/generate_dbt_schema.py
import yaml
from pathlib import Path
def contract_to_dbt_schema(contract_path: str, output_path: str):
with open(contract_path, "r") as f:
contract = yaml.safe_load(f)
schema_name = contract["dataset"]["name"]
fields = contract["schema"]["fields"]
columns = []
for field in fields:
col = {
"name": field["name"],
"description": field.get("description", ""),
"tests": [],
}
tests = field.get("tests", []) or []
for test in tests:
if test == "not_null":
col["tests"].append("not_null")
elif test == "accepted_values":
col["tests"].append({
"accepted_values": {
"values": field["allowed_values"]
}
})
elif isinstance(test, dict) and "matches_regex" in test:
# dbt-utils regex test
col["tests"].append({
"dbt_utils.expression_is_true": {
"expression": f"REGEXP_LIKE({field['name']}, '{test['matches_regex']}')"
}
})
# Uniqueness from the contract's quality section
quality = contract.get("quality", {})
for uniqueness_rule in quality.get("uniqueness", []):
if field["name"] in uniqueness_rule["key"] and len(uniqueness_rule["key"]) == 1:
col["tests"].append("unique")
columns.append(col)
# Add composite unique tests as model-level tests
model_tests = []
for uniqueness_rule in contract.get("quality", {}).get("uniqueness", []):
if len(uniqueness_rule["key"]) > 1:
model_tests.append({
"dbt_utils.unique_combination_of_columns": {
"combination_of_columns": uniqueness_rule["key"]
}
})
dbt_schema = {
"version": 2,
"models": [
{
"name": schema_name,
"description": contract["dataset"]["description"].strip(),
"tests": model_tests,
"columns": columns,
}
],
}
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w") as f:
yaml.dump(dbt_schema, f, default_flow_style=False, sort_keys=False)
print(f"Generated dbt schema at {output_path}")
if __name__ == "__main__":
contract_to_dbt_schema(
contract_path="contracts/orders_v1.yaml",
output_path="transform/models/staging/_raw_orders.yml",
)
Beyond schema tests, you want to enforce the freshness SLA from the contract inside dbt's source freshness check. Add this to your sources.yml:
# transform/models/staging/sources.yml
version: 2
sources:
- name: raw
database: analytics_db
schema: raw
freshness:
warn_after:
count: 20
period: minute
error_after:
count: 60
period: minute
loaded_at_field: ingested_at
tables:
- name: raw_orders
description: >
Governed by data contract orders_v1.
Contract location: contracts/orders_v1.yaml
Producer: platform-engineering
And add a contract metadata macro to your staging model so downstream consumers can always trace which contract version they're operating on:
-- transform/models/staging/stg_orders.sql
{{
config(
materialized='incremental',
unique_key='order_event_id',
tags=['contract:orders', 'contract_version:1.0.0']
)
}}
with source as (
select * from {{ source('raw', 'raw_orders') }}
{% if is_incremental() %}
where ingested_at > (select max(ingested_at) from {{ this }})
{% endif %}
),
validated as (
select
-- Generate a surrogate key for this specific event
{{ dbt_utils.generate_surrogate_key(['order_id', 'status', 'event_timestamp']) }}
as order_event_id,
order_id,
customer_id,
status,
revenue_usd,
event_timestamp,
ingested_at,
-- Contract metadata — consumers can filter on this
'1.0.0' as contract_version,
current_timestamp() as transformed_at
from source
-- Soft enforcement: flag records that slipped past ingestion validation
where order_id is not null
and customer_id is not null
and status in ('PENDING', 'CONFIRMED', 'SHIPPED', 'DELIVERED', 'CANCELLED', 'REFUNDED')
)
select * from validated
Warning: Don't rely solely on dbt tests for contract enforcement. dbt tests run after data is loaded. By the time a test fails, bad data is already sitting in your staging table and may have flowed to downstream models if you're not running tests in the right order. The ingestion-time validation from the previous section is your primary defense.
Not all schema changes are equal. This is one of the most important distinctions in contract management, and it's worth spending time here because getting it wrong causes painful incidents.
A non-breaking change is one that consumers can absorb without any modification to their pipelines:
NUMERIC(10,2) to NUMERIC(12,2))A breaking change is one that forces consumers to update their code or risk failures:
NUMERIC to VARCHAR, or BIGINT to INTEGER)NOT NULL constraint to a previously nullable field)Warning about "adding enum values": Adding a new
statusvalue likePARTIALLY_REFUNDEDlooks non-breaking, but it will break any consumer code that hasCASE WHEN status = 'REFUNDED' THEN ...logic that implicitly expects to catch all refund scenarios. Treat new enum values as breaking for any downstream models that use exhaustiveCASElogic. Document this in the contract as a "soft breaking" change.
Here's a decision tree embedded in a Python script you can run against a proposed contract change to classify it automatically:
# tools/classify_contract_change.py
import yaml
import sys
from deepdiff import DeepDiff
def load_contract(path: str) -> dict:
with open(path) as f:
return yaml.safe_load(f)
def classify_change(old_contract: dict, new_contract: dict) -> dict:
old_fields = {f["name"]: f for f in old_contract["schema"]["fields"]}
new_fields = {f["name"]: f for f in new_contract["schema"]["fields"]}
breaking = []
non_breaking = []
informational = []
# Check for removed fields
for field_name in old_fields:
if field_name not in new_fields:
breaking.append(f"REMOVED FIELD: '{field_name}' exists in v{old_contract['contract']['version']} "
f"but is missing from the new version.")
# Check for added fields
for field_name in new_fields:
if field_name not in old_fields:
field = new_fields[field_name]
if not field.get("nullable", True):
breaking.append(f"ADDED NON-NULLABLE FIELD: '{field_name}' — consumers with "
f"INSERT statements or strict schema reads will fail.")
else:
non_breaking.append(f"ADDED NULLABLE FIELD: '{field_name}' — consumers can ignore this.")
# Check for modified fields
for field_name in set(old_fields) & set(new_fields):
old = old_fields[field_name]
new = new_fields[field_name]
# Type changes
if old.get("type") != new.get("type"):
breaking.append(f"TYPE CHANGE: '{field_name}' changed from "
f"'{old.get('type')}' to '{new.get('type')}'.")
# Nullability changes
if old.get("nullable", True) is True and new.get("nullable", True) is False:
breaking.append(f"NULLABILITY TIGHTENED: '{field_name}' was nullable, now non-nullable.")
elif old.get("nullable", True) is False and new.get("nullable", True) is True:
non_breaking.append(f"NULLABILITY LOOSENED: '{field_name}' was non-nullable, now nullable.")
# Enum changes
old_values = set(old.get("allowed_values") or [])
new_values = set(new.get("allowed_values") or [])
removed_values = old_values - new_values
added_values = new_values - old_values
if removed_values:
breaking.append(f"ENUM VALUES REMOVED: '{field_name}' lost values: {removed_values}.")
if added_values:
informational.append(f"ENUM VALUES ADDED: '{field_name}' gained values: {added_values}. "
f"Review consumers with exhaustive CASE logic.")
return {
"is_breaking": len(breaking) > 0,
"breaking_changes": breaking,
"non_breaking_changes": non_breaking,
"informational": informational,
}
if __name__ == "__main__":
old_path, new_path = sys.argv[1], sys.argv[2]
old = load_contract(old_path)
new = load_contract(new_path)
result = classify_change(old, new)
if result["is_breaking"]:
print("🚨 BREAKING CHANGES DETECTED — requires major version bump and migration plan:")
for c in result["breaking_changes"]:
print(f" ✗ {c}")
else:
print("✅ No breaking changes detected.")
if result["non_breaking_changes"]:
print("\nNon-breaking changes (minor version bump):")
for c in result["non_breaking_changes"]:
print(f" • {c}")
if result["informational"]:
print("\nInformational — review required:")
for c in result["informational"]:
print(f" ℹ {c}")
sys.exit(1 if result["is_breaking"] else 0)
Contracts need to evolve — the goal isn't to freeze the schema forever, it's to make changes visible, negotiated, and managed. Here's the workflow we recommend for teams using GitHub:
For non-breaking changes:
classify_contract_change.py and comments the classification on the PR1.0.0 → 1.1.0For breaking changes:
1.0.0 → 2.0.0, and the old contract moves to status: deprecated_v2 table alongside the original)Here's the CODEOWNERS file that enforces review requirements:
# .github/CODEOWNERS
# All contract files require both producer and consumer team review
contracts/ @company/platform-engineering @company/analytics-engineering
# dbt schema files generated from contracts require analytics team sign-off
transform/models/staging/_*.yml @company/analytics-engineering
And the GitHub Actions CI job that runs the classifier automatically:
# .github/workflows/contract_validation.yml
name: Data Contract Validation
on:
pull_request:
paths:
- 'contracts/**'
jobs:
classify-changes:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: pip install pyyaml deepdiff
- name: Find changed contract files
id: changed-contracts
run: |
git diff --name-only origin/main...HEAD -- 'contracts/*.yaml' > changed_contracts.txt
cat changed_contracts.txt
- name: Classify contract changes
run: |
while IFS= read -r contract_file; do
echo "Checking: $contract_file"
old_version=$(git show origin/main:"$contract_file" 2>/dev/null || echo "")
if [ -z "$old_version" ]; then
echo "New contract file — no classification needed."
else
git show origin/main:"$contract_file" > /tmp/old_contract.yaml
python tools/classify_contract_change.py /tmp/old_contract.yaml "$contract_file"
fi
done < changed_contracts.txt
- name: Regenerate dbt schemas from contracts
run: |
while IFS= read -r contract_file; do
contract_id=$(python -c "import yaml; c=yaml.safe_load(open('$contract_file')); print(c['contract']['id'])")
python tools/generate_dbt_schema.py
done < changed_contracts.txt
- name: Check for uncommitted dbt schema changes
run: |
git diff --exit-code transform/models/staging/_*.yml || \
(echo "ERROR: dbt schema files are out of sync with contracts. Run generate_dbt_schema.py and commit." && exit 1)
Now it's your turn to put this together end-to-end. You'll implement a complete contract lifecycle for a user_events dataset.
Scenario: Your product team's mobile app emits user behavior events (page views, button clicks, purchases) via a webhook receiver. The analytics team needs to build a funnel analysis pipeline on top of this data. They've been burned before when the event schema changed without warning. You're going to fix that.
Step 1: Write the initial contract
Create contracts/user_events_v1.yaml. Your contract must include:
event_id, user_id, session_id, event_type, page_name, and occurred_atevent_type should have an allowed_values list with at least 4 values[event_id, occurred_at]Step 2: Write a validation script
Using the contract_validator.py pattern from this lesson, write a script that:
user_events_v1.yamlevent_type value, null user_id)Step 3: Simulate a breaking change
Create contracts/user_events_v2_proposed.yaml where:
page_name is renamed to screen_name (breaking)platform is added with allowed values ['ios', 'android', 'web'] (non-breaking)Run classify_contract_change.py against v1 and v2_proposed. Confirm it correctly identifies the breaking change and classifies the non-breaking addition.
Step 4: Write the migration plan
In a file called contracts/migrations/user_events_v1_to_v2.md, document:
Step 5: Generate and inspect the dbt schema
Adapt generate_dbt_schema.py to work with your user_events_v1.yaml contract. Run it and inspect the output. Check that the accepted_values test for event_type appears correctly in the generated YAML.
Mistake 1: Treating the contract as documentation rather than enforcement
The most common failure mode. Teams write a contract, put it in Confluence, and then nobody checks it when making changes. Contracts only work when they're machine-readable, version-controlled, and enforced in CI. If you can't run a script against your contract to validate data, you don't have a contract — you have a wiki page.
Mistake 2: Only enforcing at the dbt layer
If your only validation is dbt tests, you're catching violations after data is already loaded. By then, it may have cascaded into multiple downstream models. Always enforce as early as possible — ideally at ingestion time before any write to persistent storage.
Mistake 3: Blanket rejection of all invalid events
Rejecting 100% of a batch because 2% of events are invalid destroys availability. Use a threshold policy: below 2% violations, quarantine and continue; above 2%, halt and alert. Define the threshold in the contract itself so it's explicit and agreed upon.
Mistake 4: Not including semantics in the contract
A field named revenue_usd that is sometimes NULL for cancelled orders is not obvious to every consumer. If you only define the schema and skip the semantic documentation, you'll have consumers writing incorrect SQL that looks correct. The description field in your contract is not optional filler — it's where bugs live.
Mistake 5: Version numbers without migration plans
Bumping to v2.0.0 is meaningless if consumers don't know what changed, when v1 is going away, or what they need to do. Every major version bump must come with a written migration document committed alongside the contract change. Make this a PR merge requirement in your CODEOWNERS config.
Mistake 6: Forgetting about the additionalProperties decision
In your JSON Schema validator, additionalProperties: true means producers can add new fields without breaking validation. This is usually what you want (it makes field additions non-breaking at the ingestion layer), but you should explicitly decide this rather than accidentally defaulting to one or the other. Document your team's policy in the contract's on_additional_fields key.
Troubleshooting: dbt tests passing but data is wrong
If your dbt tests pass but downstream dashboards are showing incorrect numbers, check whether your tests are actually covering the right rows. An incremental model that only tests newly loaded rows won't catch existing corruption. Periodically run a full-refresh test run (not in production — use a clone) to validate the full dataset against the contract.
Troubleshooting: Classifier giving false positives on type changes
If you're using Snowflake-specific types like TIMESTAMP_TZ and your classifier flags them as changed when nothing actually changed, it's because YAML is loading the type strings differently (e.g., trailing whitespace, aliases). Normalize all type strings to uppercase and strip whitespace in your load_contract function before comparison.
You've now built a complete data contract implementation that covers all four stages of the lifecycle:
The core discipline here is treating data interfaces the same way software engineers treat API interfaces: as explicit, versioned contracts between teams that require negotiation to change, not silent assumptions.
Where to go from here: