Implementing Data Mesh at Scale: Architecture and Governance

# Data Product Example: Customer 360
# Implemented using dbt (data build tool)

# models/customer_360/schema.yml
version: 2

models:
  - name: customer_360
    description: >
      Unified customer view combining profile, activity, and support data.
      Owner: customer-analytics@company.com
      SLA: Daily refresh by 8am UTC
      Quality: 99% completeness, <1% duplicates
    
    meta:
      owner: customer-analytics@company.com
      domain: customer
      sla_freshness_hours: 24
      quality_score_target: 0.99
      
    columns:
      - name: customer_id
        description: Unique customer identifier
        tests:
          - unique
          - not_null
      
      - name: email
        description: Customer email address
        tests:
          - unique
          - not_null
      
      - name: first_order_date
        description: Date of customer's first order
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_of_type:
              column_type: date
      
      - name: total_revenue
        description: Lifetime customer revenue
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 1000000
      
      - name: last_activity_date
        description: Most recent customer activity
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: "2020-01-01"
              max_value: "{{ var('current_date') }}"

# models/customer_360/customer_360.sql
{{
  config(
    materialized='incremental',
    unique_key='customer_id',
    on_schema_change='fail',
    tags=['customer', 'core', 'pii'],
    meta={
      'owner': 'customer-analytics',
      'domain': 'customer'
    }
  )
}}

WITH customer_profile AS (
  SELECT
    customer_id,
    email,
    first_name,
    last_name,
    signup_date,
    country,
    segment
  FROM {{ ref('stg_customers') }}
  {% if is_incremental() %}
  WHERE updated_at >= (SELECT MAX(updated_at) FROM {{ this }})
  {% endif %}
),

order_metrics AS (
  SELECT
    customer_id,
    MIN(order_date) as first_order_date,
    MAX(order_date) as last_order_date,
    COUNT(*) as total_orders,
    SUM(order_amount) as total_revenue,
    AVG(order_amount) as avg_order_value
  FROM {{ ref('stg_orders') }}
  GROUP BY customer_id
),

activity_metrics AS (
  SELECT
    customer_id,
    MAX(event_timestamp) as last_activity_date,
    COUNT(*) as total_events,
    COUNT(DISTINCT DATE(event_timestamp)) as active_days
  FROM {{ ref('stg_events') }}
  WHERE event_timestamp >= DATEADD('day', -90, CURRENT_DATE())
  GROUP BY customer_id
),

support_metrics AS (
  SELECT
    customer_id,
    COUNT(*) as total_tickets,
    AVG(resolution_time_hours) as avg_resolution_time,
    SUM(CASE WHEN status = 'closed' THEN 1 ELSE 0 END) as resolved_tickets
  FROM {{ ref('stg_support_tickets') }}
  GROUP BY customer_id
),

final AS (
  SELECT
    p.customer_id,
    p.email,
    p.first_name,
    p.last_name,
    p.signup_date,
    p.country,
    p.segment,
    
    -- Order metrics
    COALESCE(o.first_order_date, NULL) as first_order_date,
    COALESCE(o.last_order_date, NULL) as last_order_date,
    COALESCE(o.total_orders, 0) as total_orders,
    COALESCE(o.total_revenue, 0) as total_revenue,
    COALESCE(o.avg_order_value, 0) as avg_order_value,
    
    -- Activity metrics
    COALESCE(a.last_activity_date, NULL) as last_activity_date,
    COALESCE(a.total_events, 0) as total_events_90d,
    COALESCE(a.active_days, 0) as active_days_90d,
    
    -- Support metrics
    COALESCE(s.total_tickets, 0) as total_support_tickets,
    COALESCE(s.avg_resolution_time, 0) as avg_ticket_resolution_hours,
    
    -- Computed fields
    CASE 
      WHEN o.total_orders >= 10 THEN 'champion'
      WHEN o.total_orders >= 5 THEN 'loyal'
      WHEN o.total_orders >= 2 THEN 'returning'
      WHEN o.total_orders = 1 THEN 'new'
      ELSE 'prospect'
    END as customer_lifecycle_stage,
    
    DATEDIFF('day', COALESCE(a.last_activity_date, p.signup_date), CURRENT_DATE()) as days_since_last_activity,
    
    -- Metadata
    CURRENT_TIMESTAMP() as updated_at,
    '{{ run_started_at }}' as pipeline_run_timestamp
  
  FROM customer_profile p
  LEFT JOIN order_metrics o ON p.customer_id = o.customer_id
  LEFT JOIN activity_metrics a ON p.customer_id = a.customer_id
  LEFT JOIN support_metrics s ON p.customer_id = s.customer_id
)

SELECT * FROM final

# Quality check SQL (runs after build)
# tests/customer_360/test_completeness.sql
SELECT
  COUNT(*) as total_customers,
  COUNT(CASE WHEN email IS NULL THEN 1 END) as missing_email,
  COUNT(CASE WHEN first_name IS NULL THEN 1 END) as missing_name,
  ROUND(100.0 * COUNT(CASE WHEN email IS NOT NULL THEN 1 END) / COUNT(*), 2) as completeness_pct
FROM {{ ref('customer_360') }}
HAVING completeness_pct < 99.0  -- Fail if <99% complete

# Freshness check (SLA monitoring)
# macros/check_sla.sql
{% macro check_data_freshness(model_name, max_hours) %}
  SELECT
    '{{ model_name }}' as data_product,
    MAX(updated_at) as last_update,
    DATEDIFF('hour', MAX(updated_at), CURRENT_TIMESTAMP()) as hours_since_update,
    {{ max_hours }} as sla_hours
  FROM {{ ref(model_name) }}
  HAVING hours_since_update > {{ max_hours }}
{% endmacro %}

# Run this query to check SLAs
SELECT * FROM {{ check_data_freshness('customer_360', 24) }}