Data contracts in practice: a contract file, a check before the load and Delta constraints

Part 1 tested the code on rows it made up. The code can be right and the load still wrong, because the data changed: a column added, a type changed, an order without a customer. A Data ContractData ContractAn agreement between the team that produces data and the teams that use it, written down and checked, that fixes its columns, meaning and freshness. says what the pipeline relies on, so a change like that stops the load instead of reaching a report. This part uses part 1's shop folder and module 10's files. To rebuild from nothing, run python make_days.py.

Write the contract down

The shop's pipeline relies on 4 things about the till's file: its columns in order, their types, which may never be empty, and that each order appears once. Save this as contracts/orders.yaml:

# The till's daily orders file, as the shop's pipeline relies on it.
# Owned by the till team; a change to it is agreed with the data team first.
name: till_orders
arrives: "daily by 01:00 UTC, 1 file per day: landing/orders/<day>.csv"
columns:
  - name: order_id
    type: int
    required: true
    unique: true
  - name: customer_id
    type: string
    required: true
  - name: amount
    type: decimal(10,2)
    required: true
    min: 0.01
  - name: ordered_at
    type: timestamp
    required: true

The file is short on purpose. It is the 2 teams' agreement, kept in the repository where a change to it is reviewed like code. For contracts that need more, such as owners, service levels and support channels, the Open Data Contract Standard is an open YAML format with a section for each.

Check each file against it

The check runs before the load and stops the job if the file breaks the contract. Save this as check_contract.py:

"""Check the file a run will load against its contract, before the load. Exits 1 on a breach.
Usage: python check_contract.py <run date> [contract file, contracts/orders.yaml by default]"""
import sys

import yaml
from pyspark.sql import functions as F
from delta_session import delta_spark
from shop.orders import day_for_run

path = f"landing/orders/{day_for_run(sys.argv[1])}.csv"
contract = yaml.safe_load(open(sys.argv[2] if len(sys.argv) > 2 else "contracts/orders.yaml"))
spark = delta_spark("check_contract")
columns = [column["name"] for column in contract["columns"]]
problems = []

# Spark ignores a CSV header's names when it is given a schema, so compare the header first.
header = spark.read.text(path).first().value.split(",")
if header != columns:
    problems.append(f"columns are {header}, the contract says {columns}")
else:
    schema = ", ".join(f"{column['name']} {column['type']}" for column in contract["columns"])
    orders = spark.read.option("header", True).schema(schema).csv(path)
    for column in contract["columns"]:
        name = column["name"]
        if column.get("required"):
            empty = orders.where(F.col(name).isNull()).count()
            if empty:
                problems.append(f"{empty} rows with no valid {name}")
        if column.get("unique"):
            repeated = orders.count() - orders.select(name).distinct().count()
            if repeated:
                problems.append(f"{repeated} repeated {name}")
        if "min" in column:
            low = orders.where(F.col(name) < column["min"]).count()
            if low:
                problems.append(f"{low} rows with {name} under {column['min']}")

for problem in problems:
    print(f"BREACH {path}: {problem}")
if not problems:
    print(f"ok {path}: meets the contract")
sys.exit(1 if problems else 0)

It takes a run date, like the load, and checks the file that run will load: python check_contract.py 2026-09-16 checks the file for 15 September. Then break that file 3 ways, 1 at a time: give an order a negative amount, repeat an order's line, and add a column to the header. Each breach prints a BREACH line, and the script exits with 1, which fails its task in a job. python make_days.py puts the file back.

2 details carry the check. A value that does not parse as its type, such as abc in amount, is read as null, so the required test catches bad types as well as empty cells. And the header is compared before anything else, because of how Spark reads CSV.

Why the header check comes first

Given a schema, Spark's CSV reader ignores the header's names by default (the option enforceSchema is true) and takes the columns by position. Say the till adds discount_amount between amount and ordered_at. The load would read the discount as ordered_at, fail to parse it as a timestamp, and load every order with an empty ordered_at, with no error at all. If the new column were a number placed before amount, the load would put discounts in amount, and every report built on it would be wrong.

The header check turns that into 1 clear line: the columns are not the ones agreed. The fix is then a conversation and a change to the contract, the load and its tests together, which is what the scenario the new column that only exists in development is about.

Make the table refuse bad rows

A check before the load protects 1 path into the table. The table can protect itself from every path, including a script someone runs by hand. Delta tables enforce CHECK constraints:

spark.sql("ALTER TABLE orders_daily ADD CONSTRAINT order_id_present CHECK (order_id IS NOT NULL)")
spark.sql("ALTER TABLE orders_daily ADD CONSTRAINT amount_positive CHECK (amount > 0)")

ADD CONSTRAINT first checks that every row already in the table meets it, and fails if one does not. From then on, every write is checked row by row. A write with 1 row that breaks a constraint fails whole, and since a Delta write commits all or nothing, none of its rows are added. SHOW TBLPROPERTIES orders_daily lists the constraints. Adding one upgrades the table's writer protocol, so very old Delta clients can no longer write to it. Primary key, foreign key and unique constraints on Databricks are informational: the table records them, but does not enforce them, which is why the contract check counts repeated orders itself.

Delta also enforces the table's schema on write: a DataFrame with a column the table does not have, such as discount_amount, is refused unless the write asks for schema evolution with .option("mergeSchema", "true"). That is a guard against the second half of the scenario above. Turn schema evolution on for a write when a new column has been agreed, not as a default.

Where each check goes

In the job, the contract check is a task before load_day, and load_day runs only if it succeeds, the default in Lakeflow Jobs and Airflow. The constraints sit on the table and need no task. Part 1's unit tests run before any of this ships, and part 3 watches what the contract cannot: whether the numbers still look like the shop's, day after day.