Data Contracts and Schema Validation

A data contract is an explicit agreement between a data producer and the MLOps systems that train, evaluate, or serve a model from that data. Schema validation is the executable check that enforces part of that agreement. By the end, you should be able to describe the fields, types, constraints, version rules, and failure handling for a feature dataset before it reaches training or online inference.

In data and feature pipelines, this matters because models fail quietly when input data changes. A renamed column, a new category, a unit change from dollars to cents, or a null timestamp can still produce a model artifact. Contracts make assumptions reviewable. Validators make them testable at ingestion, transformation, training, feature publication, and request serving boundaries.

What The Contract Contains

A useful contract is more than column names. It names the producer, consumer, version, owner, delivery cadence, allowed lateness, retention expectation, privacy classification, and fields downstream code may rely on. At field level it specifies presence, logical type, nullability, domain constraints, units, semantic meaning, and compatibility rules. The schema is the structural subset: names, types, required fields, nested shapes, and formats. The contract adds operational promises around that schema.

Internally, validation systems turn contract rules into predicates over records, batches, or tables. A parser loads the contract into memory. A binding layer maps contract field names to observed columns or message attributes. A type checker verifies that raw values can be interpreted as the declared logical type. Constraint evaluators test ranges, categories, uniqueness, freshness, row counts, and distribution expectations. The validator returns accepted rows, rejected rows, failing rules, severity, and identifiers for the owning producer.

Contract Anatomy

This compact contract is for a fraud model consuming payment events. It includes field rules, key definition, producer, consumer, and contract version.

columns:
  customer_id:
    type: string
    required: true
    nullable: false
  event_time:
    type: timestamp
    required: true
    nullable: false
  merchant_category:
    type: string
    required: true
    nullable: false
    allowed_values: [grocery, fuel, travel, dining, other]
  amount_usd:
    type: float
    required: true
    nullable: false
    min: 0
    max: 10000
primary_key: [customer_id, event_time]
producer: payments-events
consumer: fraud-training
version: 1.0.0

It says amount_usd is a non-null floating point value in dollars, bounded between zero and ten thousand. The category domain is closed, so crypto is rejected until the consumer updates its encoder and approves a new contract. The primary key lets training jobs and feature stores detect duplicate business events.

Example 1: Row-Level Validation

The first executable example converts the contract into simple Python rules. It demonstrates required fields, coercion, category checks, and numeric bounds. A production stack may generate similar checks from Avro, Protocol Buffers, JSON Schema, Great Expectations, Pandera, or a feature platform.

from datetime import datetime

rules = {
    "customer_id": (str, None),
    "event_time": (datetime, None),
    "merchant_category": (str, {"grocery", "fuel", "travel", "dining", "other"}),
    "amount_usd": (float, (0, 10000)),
}

def coerce(value, kind):
    if kind is datetime:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    return kind(value)

def validate(row):
    errors = []
    for field, (kind, constraint) in rules.items():
        if field not in row or row[field] is None:
            errors.append((field, "missing or null"))
            continue
        try:
            value = coerce(row[field], kind)
        except ValueError:
            errors.append((field, "bad type"))
            continue
        if isinstance(constraint, set) and value not in constraint:
            errors.append((field, "bad category"))
        if isinstance(constraint, tuple) and not constraint[0] <= value <= constraint[1]:
            errors.append((field, "out of range"))
    return errors

print(validate({"customer_id": "c-100", "event_time": "2026-09-06T10:00:00Z", "merchant_category": "fuel", "amount_usd": "42.10"}))
print(validate({"customer_id": "c-101", "event_time": "bad", "merchant_category": "crypto", "amount_usd": "-3.00"}))

The expected output begins with an empty list for the valid row. The invalid row returns three errors: event_time has a bad type, merchant_category has a bad category, and amount_usd is out of range. Validation happens before feature computation, preventing contaminated rows from changing training statistics or encoders.

Example 2: Compatibility Rules

Contracts need version semantics because producers and consumers rarely deploy together. A compatible change is one old consumers can ignore safely, such as adding an optional column. A breaking change removes a required field, changes a logical type, changes units without a new field, narrows a domain, or alters event meaning.

producer = {
    "customer_id": "string",
    "event_time": "timestamp",
    "merchant_category": "string",
    "amount_usd": "float",
    "device_id": "string",
}
consumer = {
    "customer_id": "string",
    "event_time": "timestamp",
    "merchant_category": "string",
    "amount_usd": "float",
}

def classify(old, new):
    breaking = [f"removed {k}" for k in old if k not in new]
    breaking += [f"changed {k}" for k in old if k in new and old[k] != new[k]]
    compatible = [f"added optional {k}" for k in new if k not in old]
    return {"breaking": breaking, "compatible": compatible}

print(classify(consumer, producer))

The output reports device_id as a compatible optional addition. If the producer changed amount_usd from float to string, the function would mark the release as breaking. In an MLOps pipeline, that should block automatic promotion until feature code, training code, and evaluation baselines are updated together.

Example 3: Batch Gate For Training Data

Row checks catch malformed records, but training pipelines also need batch gates. A batch gate decides whether a run should proceed, continue with quarantined rows, or fail closed. For a fraud model, a few unknown categories can be quarantined; a missing amount column should stop the run because the trained model would represent a different problem.

import csv
from pathlib import Path

rows = [
    ["c-1", "grocery", "18.25"],
    ["c-2", "travel", "812.40"],
    ["c-3", "unknown", "7.00"],
]
path = Path("events.csv")
path.write_text("customer_id,merchant_category,amount_usd\n" + "\n".join(",".join(row) for row in rows))
allowed = {"grocery", "fuel", "travel", "dining", "other"}
accepted, rejected = 0, []
for line, row in enumerate(csv.DictReader(path.open()), start=2):
    amount = float(row["amount_usd"])
    if row["merchant_category"] not in allowed:
        rejected.append(f"line {line}: merchant_category")
    elif amount < 0 or amount > 10000:
        rejected.append(f"line {line}: amount_usd")
    else:
        accepted += 1
print(f"accepted={accepted}")
print("rejected=" + ", ".join(rejected))
path.unlink()

This script writes a tiny CSV, validates each row, reports accepted=2, reports the rejected line, and removes the temporary file. One row is quarantined because unknown is outside the allowed category set. The training job can now fail above a threshold, attach samples to an incident, or request a contract update.

Design Choices And Trade-Offs

Choose validation placement deliberately. Producer-side validation catches errors closest to the source and gives fast feedback. Consumer-side validation protects each model pipeline from assumptions the producer may not know. Many systems use both: producer checks enforce the published contract, while consumer checks enforce model-specific constraints such as feature availability and label window rules.

Choose strictness by consequence. Fail-closed policies fit labels, sensitive fields, joins that can duplicate rows, and features whose unit changes would invert model behavior. Fail-open with quarantine can fit optional enrichment fields or a small number of malformed events in high-volume streams. The trade-off is availability versus correctness: continuing keeps pipelines moving, but it can hide upstream degradation unless rejected counts and rule names are monitored.

Choose schema technology according to boundary. Message buses often use Avro or Protocol Buffers. REST payloads often use JSON Schema or OpenAPI. Dataframe-heavy training code may use Pandera-like checks. Warehouse tables may use SQL constraints or expectation suites. The tool is less important than keeping the same contract identity visible in lineage metadata, model registry entries, and deployment records.

Failure Modes And Troubleshooting

Symptom: a nightly training job fails with missing column errors. Cause: the producer renamed merchant_category to merchant_type without publishing a breaking contract version. Diagnostic steps: compare the observed table schema with the stored contract, inspect producer deployment history, and query the first batch where the column disappeared. Correction: roll back the producer, add a compatibility alias, or release a coordinated consumer update.

Symptom: validation passes, but model quality drops after retraining. Cause: the schema remained stable while semantics changed, for example amount_usd began arriving in cents. Diagnostic steps: inspect distribution summaries, compare percentiles against previous batches, and sample raw events. Correction: add a unit rule or distribution expectation, fix the producer mapping, and retrain only after corrected data is available.

Symptom: online inference returns many default predictions. Cause: the serving validator rejects a new category that the request path maps to an all-zero feature vector. Diagnostic steps: check rejection counters by rule, inspect feature encoding logs, and compare online categories with the training vocabulary. Correction: quarantine or map the category through an explicit unknown bucket, update the contract if the category is real, and retrain if needed.

Security, Reliability, And Performance

Contracts reduce security risk by marking sensitive fields and preventing accidental propagation into feature tables or logs. A validator should report field names and rule identifiers, not full payloads containing customer data. They improve reliability by giving orchestrators a clear failure boundary: a rejected batch should not register a model candidate, publish online features, or overwrite accepted training data. For performance, validate in the cheapest place that still protects the boundary. Streaming systems may sample expensive distribution checks while applying cheap type and required-field checks to every event.

Hands-On Lab

Prerequisites: Python with the standard library is enough. Create an empty working directory so the temporary CSV file cannot collide with production data. No external services are required.

  1. Copy the batch gate example into validate_events.py.
  2. Run python validate_events.py and confirm it prints accepted=2 followed by one rejected line.
  3. Change the third row category from unknown to dining. Run again and confirm all three rows are accepted.
  4. Change one amount_usd value to -1.00. Run again and confirm the amount rule rejects that row.
  5. Add a fourth row with an extra field. Decide whether extra fields are compatible, ignored, or rejected, then update the script to enforce that decision.

Verification: keep expected output beside the test data and automate the command in your pipeline. A contract change should update both the contract file and expected validator result. Cleanup: the example removes events.csv. If interrupted before cleanup, delete that local file and rerun from a clean directory.

Assessment Exercises

  1. A producer wants to change amount_usd to an integer number of cents. Is this compatible, breaking, or compatible only with a new field? Explain the model risk.
  2. Where would you place validation for a feature used by both training and online serving, and what different checks would run at each boundary?
  3. Your validator rejects 4 percent of rows because of a new category. Define a fail-open policy and a fail-closed policy, then state which metrics would reveal model harm.
  4. Write a contract rule that detects duplicate payment events. What key would you use, and what false positives could occur?
  5. How would you record contract version, validation result, and rejected-row count so a model registry entry can be traced back to the data it used?

Summary

Data contracts make feature assumptions explicit, and schema validation turns those assumptions into executable gates. The important mechanics are field-level rules, compatibility classification, validation placement, structured failure output, and lineage. In MLOps, these checks protect training data, feature stores, model promotion, and serving requests from silent data changes ordinary code tests rarely catch.