Data Mesh in de Praktijk: Gedistribueerd Eigenaarschap

Wat is Data Mesh — en waarom is het in 2026 relevanter dan ooit?

Grote organisaties lopen al jaren tegen hetzelfde probleem aan: een centraal data-team dat de bottleneck is voor alles. Elke rapportage, elk dashboard, elke ML-pipeline moet door dezelfde poort. De backlog groeit, de frustratie ook, en de datakwaliteit lijdt er onder omdat het centrale team simpelweg niet genoeg domeinkennis heeft om alle data goed te begrijpen.

Data Mesh is het architectuurparadigma dat dit fundamenteel probeert op te lossen. In plaats van alle data naar één centraal platform te trechteren, geeft Data Mesh het eigenaarschap van data terug aan de domeinen die die data het beste kennen. Het concept werd in 2019 geïntroduceerd door Zhamak Dehghani en is sindsdien uitgegroeid tot een van de meest besproken architectuurbenaderingen in de data-wereld.

In 2026 zien we dat steeds meer Nederlandse organisaties — van banken en verzekeraars tot overheidsinstanties en scale-ups — actief aan de slag gaan met Data Mesh. De druk vanuit AVG-compliance, de groeiende complexiteit van datalandschappen en de behoefte aan snellere time-to-insight maken gedecentraliseerd eigenaarschap niet langer optioneel, maar noodzakelijk.

De vier pijlers van Data Mesh

Data Mesh is geen technologie maar een architectuurfilosofie. Om het te begrijpen, moet je de vier fundamentele principes kennen die samen het fundament vormen.

Hoe werkt Data Mesh in de praktijk?

Laten we de architectuur stap voor stap doorlopen aan de hand van een realistisch scenario: een Nederlandse e-commerce organisatie met domeinen als Orders, Klanten, Logistiek en Marketing.

Praktische implementatie: codevoorbeelden

Laten we de theorie vertalen naar concrete code. We gebruiken Python, dbt en YAML om een realistisch data-product te definiëren en te implementeren binnen een Data Mesh architectuur op Databricks / Delta Lake.

1. Data Product Manifest (YAML contract)

Elk data-product begint met een machine-leesbaar manifest. Dit fungeert als het contract tussen producent en consument.

# data-product.yaml — Domein: Orders
apiVersion: dataproduct/v1
kind: DataProduct
metadata:
  name: orders-daily-summary
  domain: orders
  owner: team-orders@acme.nl
  version: "2.1.0"
  tags:
    - financial
    - pii: false
    - classification: internal

spec:
  description: >
    Dagelijkse samenvatting van alle orders per klant en productcategorie.
    Vernieuwd elke nacht om 03:00 CET. Bevat geen PII-data.

  sla:
    freshness: "dagelijks, uiterlijk 06:00 CET"
    availability: "99.5%"
    support_contact: "data-orders@acme.nl"

  output_ports:
    - type: delta_table
      location: "catalog.orders_domain.orders_daily_summary"
      format: delta
      partitioned_by: ["order_date"]
    - type: rest_api
      endpoint: "https://data-api.acme.nl/orders/daily-summary"
      auth: oauth2

  quality_checks:
    - name: no_null_order_ids
      rule: "COUNT(*) WHERE order_id IS NULL = 0"
    - name: daily_order_count_minimum
      rule: "COUNT(*) WHERE order_date = CURRENT_DATE >= 1000"
    - name: revenue_positive
      rule: "MIN(total_revenue) >= 0"

  lineage:
    upstream:
      - source_system: "SAP-ERP"
        table: "raw.sap.sales_orders"
      - source_system: "Shopify"
        table: "raw.shopify.orders"
    downstream_consumers:
      - domain: marketing
        contact: team-marketing@acme.nl
      - domain: finance
        contact: team-finance@acme.nl

2. dbt model voor het data-product

Het orders-domein gebruikt dbt om de transformatielogica te beheren. Het model bevat ingebouwde datakwaliteitschecks.

-- models/orders_domain/orders_daily_summary.sql
{{
  config(
    materialized='incremental',
    unique_key='order_date || \'-\' || customer_segment',
    partition_by={
      "field": "order_date",
      "data_type": "date",
      "granularity": "day"
    },
    tags=['data_product', 'domain_orders', 'daily'],
    meta={
      'owner': 'team-orders@acme.nl',
      'data_product': 'orders-daily-summary',
      'sla_freshness_hours': 6
    }
  )
}}

WITH raw_orders AS (
    SELECT
        order_id,
        customer_id,
        order_date,
        product_category,
        quantity,
        unit_price,
        quantity * unit_price AS line_revenue,
        order_status,
        channel
    FROM {{ ref('stg_orders') }}
    WHERE order_status NOT IN ('cancelled', 'test')

    {% if is_incremental() %}
      AND order_date >= (SELECT MAX(order_date) - INTERVAL '3 days' FROM {{ this }})
    {% endif %}
),

customer_segments AS (
    SELECT
        customer_id,
        customer_segment,
        is_premium
    FROM {{ ref('dim_customers') }}
),

aggregated AS (
    SELECT
        o.order_date,
        cs.customer_segment,
        o.product_category,
        o.channel,
        COUNT(DISTINCT o.order_id)   AS total_orders,
        COUNT(DISTINCT o.customer_id) AS unique_customers,
        SUM(o.line_revenue)           AS total_revenue,
        AVG(o.line_revenue)           AS avg_order_value,
        SUM(o.quantity)               AS total_units_sold
    FROM raw_orders o
    LEFT JOIN customer_segments cs USING (customer_id)
    GROUP BY 1, 2, 3, 4
)

SELECT
    *,
    CURRENT_TIMESTAMP() AS _data_product_updated_at,
    '2.1.0'             AS _data_product_version
FROM aggregated

3. Python: data-product kwaliteitsvalidatie

Een generiek validatiescript dat het YAML manifest inleest en automatisch de gedefinieerde kwaliteitschecks uitvoert via PySpark of de Databricks SQL API.

"""
data_product_validator.py
Valideert een data-product op basis van zijn manifest.
Gebruik: python data_product_validator.py --manifest data-product.yaml
"""

import yaml
import argparse
from databricks import sql
from datetime import datetime
from dataclasses import dataclass, field
from typing import List

@dataclass
class ValidationResult:
    check_name: str
    passed: bool
    message: str
    timestamp: datetime = field(default_factory=datetime.utcnow)

class DataProductValidator:
    def __init__(self, manifest_path: str, db_connection):
        with open(manifest_path, 'r') as f:
            self.manifest = yaml.safe_load(f)
        self.conn = db_connection
        self.results: List[ValidationResult] = []

    def get_delta_table(self) -> str:
        for port in self.manifest['spec']['output_ports']:
            if port['type'] == 'delta_table':
                return port['location']
        raise ValueError("Geen Delta table output port gevonden")

    def run_quality_checks(self) -> List[ValidationResult]:
        table = self.get_delta_table()
        checks = self.manifest['spec'].get('quality_checks', [])

        for check in checks:
            check_name = check['name']
            rule = check['rule']

            try:
                # Evalueer de regel als een SQL-expressie
                query = f"""
                    SELECT CASE
                        WHEN ({rule}) THEN 'PASS'
                        ELSE 'FAIL'
                    END AS result
                    FROM {table}
                    HAVING COUNT(*) > 0
                """
                cursor = self.conn.cursor()
                cursor.execute(query)
                row = cursor.fetchone()
                passed = row[0] == 'PASS' if row else False

                self.results.append(ValidationResult(
                    check_name=check_name,
                    passed=passed,
                    message=f"Check '{check_name}': {'✅ PASS' if passed else '❌ FAIL'}"
                ))
            except Exception as e:
                self.results.append(ValidationResult(
                    check_name=check_name,
                    passed=False,
                    message=f"Check '{check_name}': ❌ ERROR — {str(e)}"
                ))

        return self.results

    def validate_freshness(self) -> ValidationResult:
        """Controleer of de data vers genoeg is op basis van SLA."""
        table = self.get_delta_table()
        sla_hours = 6  # Default; idealiter uit manifest lezen

        query = f"""
            SELECT
                MAX(_data_product_updated_at) AS last_update,
                TIMESTAMPDIFF(HOUR, MAX(_data_product_updated_at), NOW()) AS hours_since_update
            FROM {table}
        """
        cursor = self.conn.cursor()
        cursor.execute(query)
        row = cursor.fetchone()

        if row and row[1] is not None:
            hours_stale = row[1]
            passed = hours_stale <= sla_hours
            return ValidationResult(
                check_name="freshness_sla",
                passed=passed,
                message=f"Data is {hours_stale:.1f} uur oud (SLA: ≤{sla_hours}u): {'✅' if passed else '❌'}"
            )

        return ValidationResult(
            check_name="freshness_sla",
            passed=False,
            message="Kon versheid niet bepalen"
        )

    def print_report(self):
        product_name = self.manifest['metadata']['name']
        domain = self.manifest['metadata']['domain']
        print(f"\n{'='*60}")
        print(f"Data Product Validatierapport")
        print(f"Product : {product_name}")
        print(f"Domein  : {domain}")
        print(f"Tijdstip: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC")
        print(f"{'='*60}")

        passed = sum(1 for r in self.results if r.passed)
        total  = len(self.results)

        for result in self.results:
            print(f"  {result.message}")

        print(f"\nResultaat: {passed}/{total} checks geslaagd")
        print(f"Status   : {'✅ HEALTHY' if passed == total else '❌ DEGRADED'}")
        print(f"{'='*60}\n")

        return passed == total


# ---- Gebruik ----
if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('--manifest', required=True)
    parser.add_argument('--server-hostname', required=True)
    parser.add_argument('--http-path', required=True)
    parser.add_argument('--access-token', required=True)
    args = parser.parse_args()

    connection = sql.connect(
        server_hostname=args.server_hostname,
        http_path=args.http_path,
        access_token=args.access_token
    )

    validator = DataProductValidator(args.manifest, connection)
    validator.run_quality_checks()
    validator.validate_freshness()
    success = validator.print_report()

    exit(0 if success else 1)

4. Terraform: domein-geïsoleerde infrastructuur

Met Terraform zorgt het platform-team dat elk domein een geïsoleerde omgeving krijgt in Unity Catalog, inclusief correcte access control.

# modules/data_domain/main.tf
# Herbruikbaar module dat het platform-team aanbiedt aan elk domein

variable "domain_name"  { type = string }
variable "owner_group"  { type = string }
variable "catalog_name" { type = string }

resource "databricks_schema" "domain_schema" {
  catalog_name = var.catalog_name
  name         = "${var.domain_name}_domain"
  comment      = "Schema eigendom van domein: ${var.domain_name}"

  properties = {
    domain  = var.domain_name
    owner   = var.owner_group
    managed = "data-mesh"
  }
}

# Domein-team krijgt volledige rechten op eigen schema
resource "databricks_grants" "domain_owner_grants" {
  schema = databricks_schema.domain_schema.id

  grant {
    principal  = var.owner_group
    privileges = ["ALL_PRIVILEGES"]
  }
}

# Andere domeinen krijgen alleen SELECT op output-poorten
resource "databricks_grants" "consumer_read_grants" {
  for_each = toset(var.consumer_groups)
  schema   = databricks_schema.domain_schema.id

  grant {
    principal  = each.value
    privileges = ["SELECT", "USE_SCHEMA"]
  }
}

Data Mesh vs. alternatieven

Data Mesh is niet de enige aanpak. Afhankelijk van je organisatiegrootte en volwassenheid kunnen andere architectuurpatronen beter passen. Hier is een eerlijke vergelijking:

Praktijkcase: Nederlandse retailer implementeert Data Mesh

Best practices voor Data Mesh in productie