Data Mesh Architecture: Decentralized Data Ownership at Scale

# data-product.yaml
apiVersion: datamesh/v1
kind: DataProduct
metadata:
  name: customer-360
  domain: sales
  owner: sales-data-team
  version: 2.1.0

spec:
  description: |
    Unified customer view combining CRM, transaction,
    and behavioral data for comprehensive customer analytics.

  classification:
    type: aggregate
    tier: gold
    pii: true

  inputPorts:
    - name: crm-customers
      type: cdc-stream
      source: salesforce
      format: avro

    - name: transactions
      type: batch
      source: erp-system
      format: parquet
      schedule: "0 */4 * * *"

    - name: web-events
      type: stream
      source: segment
      format: json

  outputPorts:
    - name: sql-access
      type: sql
      engine: trino
      catalog: customer_analytics
      schema: customer_360

    - name: api-access
      type: rest
      endpoint: /api/v2/customers
      authentication: oauth2

    - name: stream-access
      type: kafka
      topic: customer.360.events
      format: avro

  schema:
    format: iceberg
    location: s3://data-lake/customer-360/
    partitionBy:
      - region
      - year(updated_at)
    sortBy:
      - customer_id

  slos:
    freshness:
      target: 4h
      measurement: last_updated_timestamp
    completeness:
      target: 99.5%
      measurement: non_null_ratio
    accuracy:
      target: 99.9%
      measurement: validation_pass_rate
    availability:
      target: 99.9%
      measurement: uptime_percentage

  governance:
    dataClassification: confidential
    retentionPeriod: 7y
    gdprCompliant: true
    accessControl:
      - role: sales-analysts
        permissions: [read]
      - role: data-scientists
        permissions: [read]
      - role: sales-data-team
        permissions: [read, write, admin]

# governance_policies.py
from datamesh.governance import Policy, PolicyEngine, DataProduct

class InteroperabilityPolicy(Policy):
    """Global policy ensuring data product interoperability."""

    def validate(self, data_product: DataProduct) -> bool:
        checks = [
            self.check_schema_registry(data_product),
            self.check_naming_conventions(data_product),
            self.check_documentation(data_product),
            self.check_output_ports(data_product),
        ]
        return all(checks)

    def check_schema_registry(self, dp: DataProduct) -> bool:
        """All schemas must be registered in central schema registry."""
        return dp.schema.is_registered()

    def check_naming_conventions(self, dp: DataProduct) -> bool:
        """Domain.product.version naming convention."""
        pattern = r"^[a-z]+\.[a-z-]+\.v\d+$"
        return re.match(pattern, dp.qualified_name) is not None

    def check_documentation(self, dp: DataProduct) -> bool:
        """Minimum documentation requirements."""
        required_fields = ['description', 'owner', 'slos', 'schema']
        return all(hasattr(dp.metadata, f) for f in required_fields)

    def check_output_ports(self, dp: DataProduct) -> bool:
        """At least one standardized output port required."""
        standard_ports = ['sql', 'rest', 'graphql', 'kafka']
        return any(p.type in standard_ports for p in dp.output_ports)


class PIIHandlingPolicy(Policy):
    """Policy for PII data handling compliance."""

    PII_COLUMNS = ['email', 'ssn', 'phone', 'address', 'dob', 'name']

    def validate(self, data_product: DataProduct) -> bool:
        if not data_product.contains_pii:
            return True

        return (
            self.check_pii_encryption(data_product) and
            self.check_pii_masking(data_product) and
            self.check_gdpr_compliance(data_product) and
            self.check_access_controls(data_product)
        )


# Policy Engine Configuration
policy_engine = PolicyEngine(
    policies=[
        InteroperabilityPolicy(severity='error'),
        PIIHandlingPolicy(severity='error'),
    ],
    enforcement_mode='strict',
)

# data-contract.yaml
apiVersion: datacontract/v1
kind: DataContract
metadata:
  name: customer-transactions-contract
  version: 1.2.0
  owner: sales-data-team

producer:
  domain: sales
  dataProduct: customer-transactions
  team: sales-data-team
  contact: sales-data@company.com

consumers:
  - domain: marketing
    team: marketing-analytics
    useCase: Campaign attribution analysis

  - domain: finance
    team: financial-reporting
    useCase: Revenue recognition

schema:
  type: avro
  specification: |
    {
      "type": "record",
      "name": "CustomerTransaction",
      "namespace": "com.company.sales",
      "fields": [
        {"name": "transaction_id", "type": "string"},
        {"name": "customer_id", "type": "string"},
        {"name": "transaction_date", "type": "long", "logicalType": "timestamp-millis"},
        {"name": "amount", "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2}},
        {"name": "currency", "type": "string"},
        {"name": "channel", "type": {"type": "enum", "name": "Channel", "symbols": ["WEB", "MOBILE", "STORE", "API"]}}
      ]
    }

slos:
  freshness:
    description: Time from transaction occurrence to availability
    target: 15m
    measurement: max_latency_p99

  completeness:
    description: Percentage of transactions captured
    target: 99.9%
    measurement: captured_vs_source_count

  accuracy:
    description: Data validation pass rate
    target: 99.5%
    measurement: validation_success_rate

quality:
  rules:
    - column: transaction_id
      check: unique

    - column: amount
      check: not_null

    - column: customer_id
      check: regex
      pattern: "^CUS-[A-Z0-9]{8}$"

-- Create catalog for each domain
CREATE CATALOG IF NOT EXISTS sales_domain
COMMENT 'Sales domain data products';

CREATE CATALOG IF NOT EXISTS marketing_domain
COMMENT 'Marketing domain data products';

-- Create schema for each data product
CREATE SCHEMA IF NOT EXISTS sales_domain.customer_360
COMMENT 'Unified customer view data product'
WITH DBPROPERTIES (
  'owner' = 'sales-data-team',
  'domain' = 'sales',
  'data_product_version' = '2.1.0',
  'tier' = 'gold',
  'contains_pii' = 'true'
);

-- Create managed table with data product
CREATE TABLE IF NOT EXISTS sales_domain.customer_360.customers (
  customer_id STRING NOT NULL COMMENT 'Unique customer identifier',
  email STRING COMMENT 'Customer email (PII)',
  first_name STRING COMMENT 'Customer first name (PII)',
  last_name STRING COMMENT 'Customer last name (PII)',
  segment STRING COMMENT 'Customer segment classification',
  lifetime_value DECIMAL(12,2) COMMENT 'Customer lifetime value',
  acquisition_channel STRING COMMENT 'Original acquisition channel',
  created_at TIMESTAMP COMMENT 'Record creation timestamp',
  updated_at TIMESTAMP COMMENT 'Last update timestamp'
)
USING DELTA
PARTITIONED BY (segment)
TBLPROPERTIES (
  'delta.enableChangeDataFeed' = 'true',
  'delta.autoOptimize.optimizeWrite' = 'true'
)
COMMENT 'Customer master data with 360-degree view';

-- Set up column-level security for PII
ALTER TABLE sales_domain.customer_360.customers
ALTER COLUMN email SET MASK pii_mask;

-- Grant access to consumer groups
GRANT USAGE ON CATALOG sales_domain TO marketing_analytics;
GRANT USAGE ON SCHEMA sales_domain.customer_360 TO marketing_analytics;
GRANT SELECT ON TABLE sales_domain.customer_360.customers TO marketing_analytics;

# data_product_service.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="Data Mesh Discovery API")

class DataProductSummary(BaseModel):
    name: str
    domain: str
    description: str
    owner: str
    tier: str
    tags: List[str]
    output_ports: List[str]
    slo_status: str

@app.get("/api/v1/data-products", response_model=List[DataProductSummary])
async def list_data_products(
    domain: Optional[str] = None,
    tier: Optional[str] = None,
    owner: Optional[str] = None
):
    """List all available data products with optional filtering."""
    products = await catalog_service.list_products(
        domain=domain,
        tier=tier,
        owner=owner
    )
    return products

@app.get("/api/v1/data-products/{domain}/{product_name}")
async def get_data_product(domain: str, product_name: str):
    """Get detailed information about a specific data product."""
    product = await catalog_service.get_product(domain, product_name)
    if not product:
        raise HTTPException(status_code=404, detail="Data product not found")
    return product

@app.post("/api/v1/data-products/search")
async def search_data_products(search: SearchQuery):
    """Search data products using natural language or filters."""
    results = await catalog_service.search(
        query=search.query,
        domain=search.domain,
        tier=search.tier,
        tags=search.tags,
        limit=search.limit
    )
    return results

@app.get("/api/v1/data-products/{domain}/{product_name}/lineage")
async def get_lineage(domain: str, product_name: str, depth: int = 3):
    """Get upstream and downstream lineage for a data product."""
    lineage = await lineage_service.get_lineage(domain, product_name, depth)
    return lineage