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

A data contract writes down what a pipeline relies on in the data it receives: the columns, their types, which must be filled, which must be unique, and when it arrives. Keep it as a file in the repository, check each new file against it before loading, and stop the load on a breach. Back it with the table's own rules: Delta CHECK constraints refuse rows that break them, and schema enforcement refuses a column the table does not have.

- Page: https://tecyfy.com/data-engineering/course/testing-and-ci-cd/data-contracts
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 11: Testing, CI/CD and bundles, Data contracts, a check before the load
- Parts of this module: Unit tests for PySpark with pytest (https://tecyfy.com/data-engineering/course/testing-and-ci-cd); Data contracts, a check before the load (https://tecyfy.com/data-engineering/course/testing-and-ci-cd/data-contracts); Data quality over time (https://tecyfy.com/data-engineering/course/testing-and-ci-cd/data-quality); Declarative Automation Bundles, dev and prod (https://tecyfy.com/data-engineering/course/testing-and-ci-cd/bundles); Build it: tests and deploys in CI/CD (https://tecyfy.com/data-engineering/course/testing-and-ci-cd/build-it)

At the end of this part you can:

- Write a data contract for a source file as YAML, with its columns, types, required and unique fields
- Check a file against the contract before it is loaded, including its header, and stop the job on a breach
- Add Delta CHECK constraints so the table itself refuses rows that break the rules

[Part 1](/data-engineering/course/testing-and-ci-cd) 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 Contract](https://tecyfy.com/concepts/data-contract) 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`:

```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](https://bitol-io.github.io/open-data-contract-standard/latest/) 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`:

```python
"""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](/practice/scenarios/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**:

```python
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](/data-engineering/course/testing-and-ci-cd/data-quality) watches what the contract cannot: whether the numbers still look like the shop's, day after day.

## Check yourself

1. The till adds a column, discount_amount, between amount and ordered_at. The load reads the file with header=True and a schema of the 4 old columns. What happens by default?

   A. Spark matches the columns by their header names and ignores discount_amount
   B. The read fails because the header has 5 names
   C. Spark ignores the header's names and applies the schema by position, so discount_amount is read as ordered_at
   D. Spark adds discount_amount to the DataFrame

### Answers

1. C. With a schema given, the CSV option enforceSchema is true by default, so the header names are not checked and the columns are taken by position. The 4th column, now discount_amount, is parsed as a timestamp and fails, so ordered_at is null in every row. That is why the contract check compares the header first.

## Questions

### What is a data contract in data engineering?

A written agreement between the team that produces data and the teams that use it: the columns and their types, which must be present and unique, what the values mean, and how often and how fresh the data arrives. In practice it is a file in the repository, such as YAML, that a pipeline checks every delivery against before loading it, so a breaking change stops at the door instead of in a dashboard.

### How do I stop bad rows getting into a Delta table?

Add CHECK constraints with ALTER TABLE ... ADD CONSTRAINT, such as CHECK (amount > 0). Delta checks every new row, and a write with a row that breaks a constraint fails as a whole, so nothing from it is added. Delta also enforces the table's schema, so a write with a column the table does not have fails unless schema evolution is turned on for it.

## Sources

- [CSV Files](https://spark.apache.org/docs/latest/sql-data-sources-csv.html)
- [Constraints](https://docs.delta.io/latest/delta-constraints.html)
- [Constraints on Databricks](https://docs.databricks.com/aws/en/tables/constraints)
- [Schema enforcement](https://docs.databricks.com/aws/en/tables/schema-enforcement)
- [Definition: Open Data Contract Standard (ODCS)](https://bitol-io.github.io/open-data-contract-standard/latest/)