Data Contracts Specification

What is a data contract?

A data contract is a formal, versioned agreement between a data producer and its consumers. It defines what data will look like (schema), what it means (semantics), how fresh it will be (SLAs), and who is responsible (ownership). Contracts shift data quality left - the producer guarantees quality at the source instead of consumers building defensive checks downstream.

Contract structure

Every data contract should include these sections:

Metadata

apiVersion: datacontract/v1.0
kind: DataContract
metadata:
  name: orders                    # Unique identifier
  version: 2.1.0                  # Semantic versioning
  owner: payments-team            # Producing team
  consumers:                      # Who reads this data
    - analytics-team
    - ml-team
    - marketing-team
  tags: [payments, orders, transactional]
  description: >
    All customer orders from the checkout flow. Updated in near-real-time
    via CDC from the orders microservice PostgreSQL database.

Schema definition

schema:
  type: table
  database: warehouse
  table: curated.orders
  columns:
    - name: order_id
      type: string
      constraints: [not_null, unique]
      description: UUID v4 primary key
      pii: false

    - name: customer_id
      type: string
      constraints: [not_null]
      description: FK to customers.customer_id
      pii: true

    - name: total_amount
      type: decimal(10,2)
      constraints: [not_null]
      description: Gross order total in USD (before discounts)
      semantic: gross_revenue
      unit: USD

    - name: discount_amount
      type: decimal(10,2)
      constraints: [not_null]
      description: Total discount applied
      default: 0.00

    - name: status
      type: string
      constraints: [not_null]
      allowed_values: [pending, processing, completed, cancelled, refunded]
      description: Current order lifecycle status

    - name: created_at
      type: timestamp_tz
      constraints: [not_null]
      description: Order creation time in UTC

    - name: updated_at
      type: timestamp_tz
      constraints: [not_null]
      description: Last modification time in UTC

Semantic definitions

Avoid ambiguity by defining what business terms mean:

semantics:
  gross_revenue: >
    Total order amount before any discounts, taxes, or shipping costs.
    Includes all line items at their list price.
  net_revenue: >
    gross_revenue minus discount_amount. Does NOT include tax or shipping.
  active_order: >
    An order with status in [pending, processing]. Completed, cancelled,
    and refunded orders are not active.

SLA definitions

sla:
  freshness: 15m                  # Data must be no older than 15 minutes
  volume:
    min_rows_per_hour: 100        # Alert if fewer than 100 orders/hour
    max_rows_per_hour: 50000      # Alert if spike exceeds 50k/hour
  availability: 99.9%             # Table must be queryable 99.9% of the time
  quality:
    null_rate_threshold: 0.01     # No column should exceed 1% null rate
    duplicate_rate_threshold: 0    # Zero tolerance for duplicate order_ids

Contact and escalation

support:
  slack_channel: "#payments-data"
  oncall_rotation: payments-data-oncall
  escalation_policy: >
    1. Post in #payments-data with details
    2. If no response in 30min, page payments-data-oncall
    3. If P1 impact, page payments-eng-oncall directly

Versioning strategy

Follow semantic versioning for data contracts:

Breaking change protocol

1. Producer announces breaking change in the contract's Slack channel
2. Minimum 7-day notice period for consumer teams to adapt
3. Producer publishes new contract version (major bump)
4. Old version remains supported for a deprecation window (typically 30 days)
5. Producer removes old version after deprecation window

Enforcement patterns

Schema enforcement at ingestion

import json
import jsonschema

def validate_against_contract(record: dict, contract_path: str) -> bool:
    """Validate a single record against its data contract schema."""
    with open(contract_path) as f:
        contract = yaml.safe_load(f)

    # Convert contract columns to JSON Schema
    properties = {}
    required = []
    for col in contract["schema"]["columns"]:
        prop = {"type": _map_type(col["type"])}
        if "allowed_values" in col:
            prop["enum"] = col["allowed_values"]
        properties[col["name"]] = prop
        if "not_null" in col.get("constraints", []):
            required.append(col["name"])

    schema = {
        "type": "object",
        "properties": properties,
        "required": required,
        "additionalProperties": False,
    }

    jsonschema.validate(record, schema)
    return True

Contract testing in CI

Run contract tests on every PR that modifies a producer's schema:

# .github/workflows/contract-test.yml
name: Data Contract Tests
on:
  pull_request:
    paths:
      - 'contracts/**'
      - 'migrations/**'
      - 'models/**'

jobs:
  contract-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Validate contract schema
        run: |
          pip install datacontract-cli
          datacontract test contracts/orders-v2.yaml
      - name: Check for breaking changes
        run: |
          datacontract breaking contracts/orders-v2.yaml --against main

Runtime enforcement with Great Expectations

Generate GX expectations automatically from a data contract:

def contract_to_expectations(contract_path: str) -> list:
    """Convert a data contract YAML to Great Expectations expectations."""
    import yaml
    import great_expectations as gx

    with open(contract_path) as f:
        contract = yaml.safe_load(f)

    expectations = []
    for col in contract["schema"]["columns"]:
        if "not_null" in col.get("constraints", []):
            expectations.append(
                gx.expectations.ExpectColumnValuesToNotBeNull(column=col["name"])
            )
        if "unique" in col.get("constraints", []):
            expectations.append(
                gx.expectations.ExpectColumnValuesToBeUnique(column=col["name"])
            )
        if "allowed_values" in col:
            expectations.append(
                gx.expectations.ExpectColumnValuesToBeInSet(
                    column=col["name"],
                    value_set=col["allowed_values"],
                )
            )
    return expectations

Great Expectations - Advanced Patterns

Custom expectations

When built-in expectations don't cover your business logic, write a custom expectation by subclassing Expectation.

from great_expectations.expectations import Expectation
from great_expectations.core import ExpectationValidationResult
import re


class ExpectColumnValuesToMatchCorporateEmail(Expectation):
    """Expect column values to be valid corporate email addresses."""

    expectation_type = "expect_column_values_to_match_corporate_email"
    domain = "column"

    domain_param = "column"
    success_keys = ("column", "domain_suffix")

    default_kwarg_values = {
        "domain_suffix": "@company.com",
    }

    def _validate(self, metrics, runtime_configuration=None, execution_engine=None):
        column_values = metrics.get("column_values.nonnull")
        domain_suffix = self.kwargs.get("domain_suffix", "@company.com")

        unexpected = [
            v for v in column_values
            if not str(v).endswith(domain_suffix)
        ]

        return ExpectationValidationResult(
            success=len(unexpected) == 0,
            result={
                "unexpected_count": len(unexpected),
                "unexpected_values": unexpected[:20],
                "element_count": len(column_values),
            },
        )

Register custom expectations by placing them in a great_expectations/plugins/expectations/ directory in your project. GX auto-discovers them on context initialization.

Store backends

By default GX stores expectations and validation results on the local filesystem. For team use, configure a cloud store backend.

# great_expectations/great_expectations.yml
stores:
  expectations_store:
    class_name: ExpectationsStore
    store_backend:
      class_name: TupleS3StoreBackend
      bucket: my-gx-store
      prefix: expectations/

  validation_results_store:
    class_name: ValidationResultsStore
    store_backend:
      class_name: TupleS3StoreBackend
      bucket: my-gx-store
      prefix: validations/

  checkpoint_store:
    class_name: CheckpointStore
    store_backend:
      class_name: TupleS3StoreBackend
      bucket: my-gx-store
      prefix: checkpoints/

Supported backends: local filesystem, S3, GCS, Azure Blob, PostgreSQL.

Data Docs hosting

Data Docs are static HTML reports generated from validation results. Host them for team visibility.

# great_expectations/great_expectations.yml
data_docs_sites:
  s3_site:
    class_name: SiteBuilder
    store_backend:
      class_name: TupleS3StoreBackend
      bucket: my-data-docs
      prefix: docs/
    site_index_builder:
      class_name: DefaultSiteIndexBuilder

For internal hosting, use S3 static website hosting with CloudFront, or serve from a simple Nginx container.

Multi-batch validation

Validate expectations across multiple batches (e.g., compare today's data to yesterday's).

import great_expectations as gx

context = gx.get_context()
asset = context.data_sources.get("warehouse").get_asset("orders")

# Define batch definitions for different partitions
today_batch = asset.add_batch_definition_daily("today", column="created_at")
yesterday_batch = asset.add_batch_definition_daily("yesterday", column="created_at")

# Run same suite against both batches
suite = context.suites.get("orders_quality")

for batch_def in [today_batch, yesterday_batch]:
    validation = gx.ValidationDefinition(
        name=f"orders_{batch_def.name}",
        data=batch_def,
        suite=suite,
    )
    result = validation.run()
    print(f"{batch_def.name}: {'PASS' if result.success else 'FAIL'}")

Conditional expectations

Apply expectations only to subsets of data using row conditions.

import great_expectations as gx

suite = context.suites.get("orders_quality")

# Only check refund_amount for cancelled orders
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(
        column="refund_amount",
        row_condition='status == "cancelled"',
        condition_parser="pandas",
    )
)

# Only check shipping_address for non-digital orders
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(
        column="shipping_address",
        row_condition='product_type != "digital"',
        condition_parser="pandas",
    )
)

Integration with Airflow

Use the GreatExpectationsOperator to run checkpoints as Airflow tasks.

from airflow import DAG
from airflow.utils.dates import days_ago
from great_expectations_provider.operators.great_expectations import (
    GreatExpectationsOperator,
)

with DAG(
    dag_id="orders_quality_check",
    start_date=days_ago(1),
    schedule_interval="@daily",
    catchup=False,
) as dag:

    validate_orders = GreatExpectationsOperator(
        task_id="validate_orders",
        data_context_root_dir="/opt/airflow/great_expectations",
        checkpoint_name="orders_checkpoint",
        fail_task_on_validation_failure=True,
    )

The Airflow provider is apache-airflow-providers-great-expectations. Install it separately: pip install apache-airflow-providers-great-expectations.

Lineage Tools Comparison and Integration

Tool comparison

OpenLineage

OpenLineage is an open standard for lineage event collection. It defines a JSON schema for run events that capture inputs, outputs, and job metadata. It is not a platform - it is a protocol that platforms consume.

Core event model

RunEvent {
  eventType: START | RUNNING | COMPLETE | FAIL | ABORT
  eventTime: ISO 8601 timestamp
  run: { runId: UUID }
  job: { namespace: string, name: string }
  inputs: [ InputDataset { namespace, name, facets } ]
  outputs: [ OutputDataset { namespace, name, facets } ]
}

Native integrations

These tools emit OpenLineage events automatically when configured:

Setting up Marquez as an OpenLineage backend

Marquez is the reference implementation for collecting and serving OpenLineage events.

# Docker Compose quickstart
git clone https://github.com/MarquezProject/marquez.git
cd marquez
docker compose up -d

# Marquez API is at http://localhost:5000
# Marquez UI is at http://localhost:3000

# Point your OpenLineage clients at Marquez
from openlineage.client import OpenLineageClient

client = OpenLineageClient(url="http://localhost:5000")

Column-level lineage with facets

from openlineage.client.facet_v2 import (
    column_lineage_dataset_facet,
)

output_facets = {
    "columnLineage": column_lineage_dataset_facet.ColumnLineageDatasetFacet(
        fields={
            "total_revenue": column_lineage_dataset_facet.ColumnLineageDatasetFacetFieldsAdditional(
                inputFields=[
                    column_lineage_dataset_facet.InputField(
                        namespace="warehouse",
                        name="raw.orders",
                        field="amount",
                    ),
                    column_lineage_dataset_facet.InputField(
                        namespace="warehouse",
                        name="raw.orders",
                        field="discount",
                    ),
                ],
                transformationDescription="SUM(amount - discount)",
                transformationType="AGGREGATION",
            )
        }
    )
}

dbt lineage

dbt generates lineage automatically from ref() and source() calls in models.

Accessing lineage from the manifest

import json

with open("target/manifest.json") as f:
    manifest = json.load(f)

# Get all parents (upstream dependencies) of a model
model_key = "model.my_project.orders_summary"
node = manifest["nodes"][model_key]
parents = node["depends_on"]["nodes"]
print(f"Upstream dependencies: {parents}")

# Get all children (downstream dependents) of a model
children = manifest["child_map"].get(model_key, [])
print(f"Downstream dependents: {children}")

Column-level lineage in dbt

dbt 1.6+ supports column-level lineage in dbt Cloud. For open-source dbt, use the dbt-column-lineage package or parse SQL with sqlglot:

import sqlglot
from sqlglot.lineage import lineage

# Parse a dbt model's compiled SQL to extract column lineage
sql = """
SELECT
    o.order_id,
    o.customer_id,
    SUM(oi.quantity * oi.unit_price) AS total_amount
FROM {{ ref('orders') }} o
JOIN {{ ref('order_items') }} oi ON o.order_id = oi.order_id
GROUP BY o.order_id, o.customer_id
"""

# After compiling (replacing refs with actual table names):
compiled_sql = sql.replace("{{ ref('orders') }}", "curated.orders")
compiled_sql = compiled_sql.replace("{{ ref('order_items') }}", "curated.order_items")

result = lineage("total_amount", compiled_sql, dialect="postgres")
for node in result.walk():
    print(f"  {node.name} <- {node.source.sql()}")

DataHub

DataHub is a full metadata platform that includes lineage, data discovery, governance, and observability.

Emitting lineage to DataHub

from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import (
    DatasetLineageTypeClass,
    UpstreamClass,
    UpstreamLineageClass,
)

emitter = DatahubRestEmitter(gms_server="http://datahub-gms:8080")

upstream_lineage = UpstreamLineageClass(
    upstreams=[
        UpstreamClass(
            dataset="urn:li:dataset:(urn:li:dataPlatform:postgres,raw.orders,PROD)",
            type=DatasetLineageTypeClass.TRANSFORMED,
        ),
        UpstreamClass(
            dataset="urn:li:dataset:(urn:li:dataPlatform:postgres,raw.customers,PROD)",
            type=DatasetLineageTypeClass.TRANSFORMED,
        ),
    ]
)

emitter.emit_mce(
    {
        "proposedSnapshot": {
            "com.linkedin.pegasus2avro.metadata.snapshot.DatasetSnapshot": {
                "urn": "urn:li:dataset:(urn:li:dataPlatform:postgres,curated.order_summary,PROD)",
                "aspects": [upstream_lineage],
            }
        }
    }
)

DataHub + Airflow integration

pip install acryl-datahub-airflow-plugin

# In airflow.cfg or env vars:
export AIRFLOW__DATAHUB__DATAHUB_REST_CONN_ID=datahub_rest_default

DataHub's Airflow plugin automatically captures lineage from PostgresOperator, BigQueryOperator, and other SQL-based operators by parsing the SQL.

Choosing a lineage tool

Start with the simplest option that covers your current tools. Lineage is most valuable when it is complete - a partial lineage graph is worse than no lineage because it gives false confidence about impact analysis.