Data Contracts: Betrouwbare Pipelines in 2026

Wat zijn Data Contracts en waarom zijn ze in 2026 onmisbaar?

Stel je voor: je datawarehouse draait perfect, je dashboards zijn up-to-date en je ML-modellen produceren betrouwbare voorspellingen. Dan besluit een upstream-team stilletjes het schema van een bronsysteem aan te passen — een kolomnaam verandert, een datatype wijzigt, een veld wordt weggelaten. Binnen 24 uur vallen pipelines om, rapporten tonen nullen en je data science team belt boos op. Herkenbaar? Dit scenario speelt zich dagelijks af in organisaties wereldwijd, en data contracts zijn het antwoord.

In 2026 zijn data contracts geen nice-to-have meer. De gemiddelde enterprise-datastack bestaat uit tientallen databronnen, honderden pipelines en meerdere consumerende teams. Zonder expliciete afspraken is elke pipeline een tikkende tijdbom. Bovendien drijven regulatorische vereisten zoals DORA, AI Act en AVG organisaties richting aantoonbare datakwaliteit en traceerbaarheid — precies wat data contracts bieden.

Hoe werkt een Data Contract? Anatomie en lifecycle

Een data contract bestaat uit verschillende lagen. Het is geen simpel JSON-schemabestand, maar een levend document dat de volledige levenscyclus van data beschrijft. Hieronder vind je de anatomie van een modern data contract en de stappen in de lifecycle.

Structuur van een Data Contract (Open Data Contract Standard)

De Open Data Contract Standard (ODCS) is een opkomende industrie-standaard voor het beschrijven van data contracts in YAML. Hieronder een realistisch voorbeeld voor een orders-dataset:

# data_contract_orders_v2.yaml
# Open Data Contract Standard v2.3

apiVersion: v2.3.0
kind: DataContract
id: orders-contract-001
version: 2.1.0
status: active

info:
  title: Orders Dataset Contract
  owner: data-platform-team@bedrijf.nl
  consumer:
    - team: analytics
    - team: finance-reporting
  description: >
    Dit contract beschrijft de orders-tabel vanuit het OMS-systeem.
    Elke rij vertegenwoordigt één orderregel.
  tags:
    - commerce
    - finance
    - pii

servers:
  - environment: production
    type: bigquery
    project: dataplatform-prod
    dataset: raw_commerce
    table: orders

schema:
  - name: order_id
    type: STRING
    required: true
    unique: true
    description: Unieke identifier voor elke order
    pii: false

  - name: customer_id
    type: STRING
    required: true
    description: Koppeling naar de customers-tabel
    pii: true
    classification: internal

  - name: order_date
    type: TIMESTAMP
    required: true
    description: Tijdstip waarop de order is geplaatst
    format: "ISO 8601"

  - name: total_amount
    type: NUMERIC
    required: true
    description: Totaalbedrag van de order in EUR
    minimum: 0.01
    maximum: 999999.99

  - name: status
    type: STRING
    required: true
    description: Huidige status van de order
    enum:
      - PENDING
      - CONFIRMED
      - SHIPPED
      - DELIVERED
      - CANCELLED

  - name: created_at
    type: TIMESTAMP
    required: true

quality:
  - rule: freshness
    description: Data mag maximaal 4 uur oud zijn
    threshold: "4h"
    severity: error

  - rule: completeness
    column: customer_id
    description: customer_id mag nooit null zijn
    threshold: "100%"
    severity: error

  - rule: row_count
    description: Minimaal 1000 nieuwe rijen per dag
    minimum: 1000
    severity: warning

  - rule: uniqueness
    column: order_id
    threshold: "100%"
    severity: error

sla:
  availability: "99.9%"
  freshness: "4h"
  support_channel: "#data-platform-slack"
  incident_response: "2h"

Praktische implementatie: Soda Core en Great Expectations

Een data contract heeft alleen waarde als het wordt gehandhaafd. Hieronder laten we zien hoe je validaties implementeert met twee populaire tools: Soda Core (declaratief, YAML-gebaseerd) en Great Expectations (Python-gebaseerd, rijkere API).

Implementatie met Soda Core

Soda Core is een open-source tool die data quality checks uitvoert op basis van YAML-configuraties. Het integreert goed met Airflow, dbt en moderne data stacks.

# Installatie
pip install soda-core soda-core-bigquery

# soda_checks_orders.yml — afgeleid van het data contract
checks for orders:

  # Compleetheid: geen nulls in verplichte velden
  - missing_count(customer_id) = 0:
      name: customer_id is nooit null
      severity: error

  - missing_count(order_id) = 0:
      name: order_id is nooit null
      severity: error

  # Uniciteit
  - duplicate_count(order_id) = 0:
      name: order_id is uniek
      severity: error

  # Geldigheid van enum-waarden
  - invalid_count(status) = 0:
      name: status bevat alleen toegestane waarden
      valid values: [PENDING, CONFIRMED, SHIPPED, DELIVERED, CANCELLED]
      severity: error

  # Bereik validatie
  - min(total_amount) > 0:
      name: total_amount is altijd positief
      severity: error

  - max(total_amount) < 1000000:
      name: total_amount onder maximum
      severity: warning

  # Versheid: geen data ouder dan 4 uur
  - freshness(created_at) < 4h:
      name: Data is maximaal 4 uur oud
      severity: error

  # Volume controle
  - row_count >= 1000:
      name: Voldoende dagelijkse orders
      severity: warning

# soda_scan.py — integratie met Airflow of standalone uitvoering
from soda.scan import Scan

def run_orders_data_contract_check():
    scan = Scan()
    scan.set_data_source_name("bigquery_prod")

    scan.add_configuration_yaml_file(
        file_path="./soda_config.yml"
    )

    scan.add_sodacl_yaml_file(
        file_path="./soda_checks_orders.yml"
    )

    scan.set_scan_definition_name("orders_daily_contract_check")
    scan.set_verbose(True)

    exit_code = scan.execute()

    # Exit code 0 = succes, 2 = warnings, 3 = failures
    if exit_code == 3:
        raise ValueError(
            "❌ Data contract geschonden! "
            "Pipeline gestopt. Zie Soda Cloud voor details."
        )
    elif exit_code == 2:
        print("⚠️ Data contract warnings gedetecteerd. "
              "Pipeline gaat door maar team is genotificeerd.")

    return scan.get_scan_results()

if __name__ == "__main__":
    run_orders_data_contract_check()

Implementatie met Great Expectations

Great Expectations biedt meer flexibiliteit en programmatische controle, ideaal voor complexe validatielogica.

# great_expectations_orders_contract.py
import great_expectations as gx
from great_expectations.core.expectation_suite import ExpectationSuite

# Context initialiseren
context = gx.get_context()

# Data asset ophalen (BigQuery datasource)
datasource = context.sources.add_or_update_bigquery(
    name="bigquery_prod",
    project="dataplatform-prod",
)

data_asset = datasource.add_table_asset(
    name="orders",
    table_name="raw_commerce.orders"
)

# Expectation Suite aanmaken vanuit data contract
suite = context.add_or_update_expectation_suite("orders_contract_v2")

# --- Schema validaties ---
suite.add_expectation(gx.expectations.ExpectColumnToExist(
    column="order_id"
))
suite.add_expectation(gx.expectations.ExpectColumnToExist(
    column="customer_id"
))

# --- Compleetheid ---
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(
    column="order_id",
    meta={"contract_rule": "required field", "severity": "error"}
))
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(
    column="customer_id",
    meta={"contract_rule": "required field", "severity": "error"}
))

# --- Uniciteit ---
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(
    column="order_id",
    meta={"contract_rule": "unique constraint", "severity": "error"}
))

# --- Enum validatie ---
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeInSet(
    column="status",
    value_set=["PENDING", "CONFIRMED", "SHIPPED", "DELIVERED", "CANCELLED"],
    meta={"contract_rule": "valid enum values", "severity": "error"}
))

# --- Bereik validatie ---
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeBetween(
    column="total_amount",
    min_value=0.01,
    max_value=999999.99,
    meta={"contract_rule": "valid amount range", "severity": "error"}
))

# --- Row count ---
suite.add_expectation(gx.expectations.ExpectTableRowCountToBeGreaterThan(
    value=1000,
    meta={"contract_rule": "minimum daily volume", "severity": "warning"}
))

context.save_expectation_suite(suite)

# Checkpoint aanmaken voor CI/CD integratie
checkpoint = context.add_or_update_checkpoint(
    name="orders_contract_checkpoint",
    validations=[
        {
            "expectation_suite_name": "orders_contract_v2",
            "batch_request": data_asset.build_batch_request()
        }
    ]
)

# Uitvoeren en resultaten verwerken
results = checkpoint.run()

if not results.success:
    failed = [
        r for r in results.list_validation_results()
        if not r.success
    ]
    print(f"❌ {len(failed)} contractregels geschonden!")
    for failure in failed:
        print(f"  - {failure.expectation_config.expectation_type}: "
              f"{failure.result}")
    raise RuntimeError("Data contract validatie mislukt")

Integratie in een dbt-pipeline

# models/staging/stg_orders.yml — dbt contracten
version: 2

models:
  - name: stg_orders
    description: "Gestaged orders model — contract v2.1.0"
    config:
      contract:
        enforced: true  # dbt enforceert schema strikt

    constraints:
      - type: not_null
        columns: [order_id, customer_id, order_date]
      - type: unique
        columns: [order_id]
      - type: check
        expression: "total_amount > 0"

    columns:
      - name: order_id
        data_type: varchar
        constraints:
          - type: not_null
          - type: unique

      - name: status
        data_type: varchar
        tests:
          - accepted_values:
              values: ['PENDING', 'CONFIRMED', 'SHIPPED',
                       'DELIVERED', 'CANCELLED']

      - name: total_amount
        data_type: numeric
        tests:
          - dbt_utils.expression_is_true:
              expression: ">= 0.01"

Vergelijking: Data Contract Tools in 2026

Er zijn meerdere tools en standaarden voor data contracts. Hieronder een eerlijke vergelijking van de belangrijkste opties:

Praktijkvoorbeeld: Data Contracts bij een Retailer

Best Practices voor Production-ready Data Contracts

CI/CD Integratie met datacontract CLI

# .github/workflows/data_contract_check.yml
name: Data Contract Validation

on:
  pull_request:
    paths:
      - 'contracts/**'
      - 'models/**'

jobs:
  validate-contracts:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install datacontract CLI
        run: pip install datacontract-cli

      - name: Lint data contracts
        run: |
          datacontract lint contracts/orders_v2.yaml
          datacontract lint contracts/customers_v1.yaml

      - name: Check backward compatibility
        run: |
          # Vergelijk met vorige versie (main branch)
          git fetch origin main
          git show origin/main:contracts/orders_v2.yaml \
            > /tmp/orders_previous.yaml

          datacontract breaking \
            /tmp/orders_previous.yaml \
            contracts/orders_v2.yaml

      - name: Validate schema against live data (staging)
        env:
          GOOGLE_APPLICATION_CREDENTIALS: ${{ secrets.GCP_SA_KEY }}
        run: |
          datacontract test \
            --server staging \
            contracts/orders_v2.yaml

Conclusie: Wanneer wel en niet?

Data contracts zijn een krachtig instrument, maar geen silver bullet voor elke situatie. Hieronder een eerlijk overzicht van wanneer je ze wel en niet moet inzetten:

In 2026 is de vraag niet meer óf je data contracts moet implementeren, maar hoe snel. De tools zijn volwassen, de standaarden zijn er, en de pijn van ontbrekende contracts is goed gedocumenteerd. Begin klein: kies je meest kritieke dataset, schrijf een minimaal contract met vijf checks,