Data Mesh Architecture Pattern¶

# Data Product Manifest Example
data_product_manifest = {
    "product_name": "customer_360_view",
    "domain": "customer_experience",
    "owner": "customer-analytics-team@company.com",
    "version": "2.1.0",
    "description": "Unified customer profile with behavioral, transactional, and demographic data",

    "quality_sla": {
        "completeness": 0.95,
        "accuracy": 0.98,
        "freshness_minutes": 30,
        "availability": 0.999
    },

    "data_contract": {
        "schema_version": "1.0",
        "breaking_change_policy": "30_day_deprecation_notice",
        "backward_compatibility": True
    },

    "access": {
        "read_endpoint": "https://api.company.com/data-products/customer-360",
        "query_interface": "sql",
        "supported_formats": ["parquet", "delta", "json"],
        "authentication": "azure_ad"
    },

    "lineage": {
        "source_systems": ["crm", "web_analytics", "support_tickets"],
        "transformations": ["deduplication", "enrichment", "aggregation"],
        "last_updated": "2025-01-28T10:00:00Z"
    }
}

# Domain definition template
class DataDomain:
    """Define a data domain with clear boundaries"""

    def __init__(self, name, description, owner):
        self.name = name
        self.description = description
        self.owner = owner
        self.data_products = []
        self.source_systems = []
        self.consumers = []

    def add_data_product(self, product):
        """Register a data product in this domain"""
        self.data_products.append(product)

    def get_domain_metadata(self):
        """Return domain metadata"""
        return {
            "domain_name": self.name,
            "description": self.description,
            "owner": self.owner,
            "data_products": [p.name for p in self.data_products],
            "source_systems": self.source_systems,
            "consumer_count": len(self.consumers)
        }

# Example domain definition
customer_domain = DataDomain(
    name="customer_experience",
    description="All customer-related data products and analytics",
    owner="customer-analytics-team@company.com"
)

sales_domain = DataDomain(
    name="sales",
    description="Sales transactions, pipeline, and performance data",
    owner="sales-analytics-team@company.com"
)

from pyspark.sql import DataFrame
from delta.tables import DeltaTable
from typing import List, Dict, Any
import json

class DataProduct:
    """
    Data Product implementation with quality SLAs and governance
    """

    def __init__(self, name: str, domain: str, owner: str):
        self.name = name
        self.domain = domain
        self.owner = owner
        self.version = "1.0.0"
        self.schema_version = "1.0"
        self.sla_metrics = {}

    def define_schema(self, schema_definition: Dict):
        """Define and version the data product schema"""
        self.schema = schema_definition
        return self

    def set_quality_sla(self, metric: str, threshold: float):
        """Define quality SLA for the data product"""
        self.sla_metrics[metric] = threshold
        return self

    def publish(self, df: DataFrame, storage_path: str):
        """Publish the data product with quality checks"""

        # Validate against SLAs
        quality_checks = self._validate_quality(df)

        if not all(quality_checks.values()):
            raise ValueError(f"Quality SLA violations: {quality_checks}")

        # Add data product metadata
        df_with_metadata = df \
            .withColumn("_product_name", lit(self.name)) \
            .withColumn("_product_version", lit(self.version)) \
            .withColumn("_domain", lit(self.domain)) \
            .withColumn("_publish_timestamp", current_timestamp())

        # Write to Delta Lake
        df_with_metadata.write \
            .format("delta") \
            .mode("overwrite") \
            .option("overwriteSchema", "true") \
            .save(storage_path)

        # Register in catalog
        self._register_in_catalog(storage_path)

        return storage_path

    def _validate_quality(self, df: DataFrame) -> Dict[str, bool]:
        """Validate data against SLA metrics"""
        results = {}

        total_count = df.count()

        for metric, threshold in self.sla_metrics.items():
            if metric == "completeness":
                # Check for NULL values
                null_count = df.filter(
                    sum([col(c).isNull().cast("int") for c in df.columns]) > 0
                ).count()
                completeness = (total_count - null_count) / total_count
                results[metric] = completeness >= threshold

            elif metric == "uniqueness":
                # Check for duplicates on key columns
                distinct_count = df.dropDuplicates().count()
                uniqueness = distinct_count / total_count
                results[metric] = uniqueness >= threshold

        return results

    def _register_in_catalog(self, storage_path: str):
        """Register data product in Azure Purview"""
        catalog_entry = {
            "name": self.name,
            "domain": self.domain,
            "owner": self.owner,
            "version": self.version,
            "storage_path": storage_path,
            "schema": self.schema,
            "sla_metrics": self.sla_metrics,
            "registered_at": datetime.now().isoformat()
        }

        # Register with Purview API
        # Implementation depends on Azure Purview SDK
        pass

# Example usage
customer_product = DataProduct(
    name="customer_360_view",
    domain="customer_experience",
    owner="cx-team@company.com"
)

customer_product \
    .set_quality_sla("completeness", 0.95) \
    .set_quality_sla("uniqueness", 0.99) \
    .define_schema({
        "customer_id": "string",
        "email": "string",
        "total_orders": "integer",
        "lifetime_value": "decimal"
    })

# Build the data product
customer_df = spark.read.format("delta") \
    .load("silver/crm/customers") \
    .join(orders_df, "customer_id") \
    .groupBy("customer_id", "email") \
    .agg(
        count("order_id").alias("total_orders"),
        sum("order_amount").alias("lifetime_value")
    )

# Publish with automatic quality checks
customer_product.publish(
    customer_df,
    "data-products/customer-experience/customer-360"
)

class DataProductInterface:
    """
    Standardized interface for consuming data products
    """

    def __init__(self, product_name: str, catalog_url: str):
        self.product_name = product_name
        self.catalog_url = catalog_url
        self.metadata = self._fetch_metadata()

    def read(self, version: str = "latest") -> DataFrame:
        """Read data product with specific version"""

        storage_path = self.metadata.get("storage_path")

        if version == "latest":
            return spark.read.format("delta").load(storage_path)
        else:
            return spark.read.format("delta") \
                .option("versionAsOf", version) \
                .load(storage_path)

    def get_schema(self) -> Dict:
        """Get data product schema"""
        return self.metadata.get("schema")

    def get_sla(self) -> Dict:
        """Get quality SLA metrics"""
        return self.metadata.get("sla_metrics")

    def get_lineage(self) -> Dict:
        """Get data product lineage"""
        # Query Purview for lineage information
        pass

    def _fetch_metadata(self) -> Dict:
        """Fetch metadata from catalog"""
        # Implementation depends on catalog API
        pass

# Consumer usage
customer_360 = DataProductInterface(
    product_name="customer_360_view",
    catalog_url="https://purview.azure.com/catalog"
)

# Read latest version
df = customer_360.read()

# Check schema
schema = customer_360.get_schema()
print(f"Schema: {schema}")

# Check SLAs
sla = customer_360.get_sla()
print(f"Quality SLA: {sla}")

# Terraform configuration for data product infrastructure
"""
resource "azurerm_synapse_workspace" "domain_workspace" {
  name                = "${var.domain_name}-synapse"
  resource_group_name = azurerm_resource_group.rg.name
  location            = azurerm_resource_group.rg.location

  storage_data_lake_gen2_filesystem_id = azurerm_storage_data_lake_gen2_filesystem.domain_storage.id

  sql_administrator_login          = "sqladmin"
  sql_administrator_login_password = var.sql_admin_password

  managed_virtual_network_enabled = true

  identity {
    type = "SystemAssigned"
  }

  tags = {
    domain = var.domain_name
    owner  = var.domain_owner
  }
}

resource "azurerm_synapse_spark_pool" "domain_spark" {
  name                 = "${var.domain_name}-spark"
  synapse_workspace_id = azurerm_synapse_workspace.domain_workspace.id

  node_size_family = "MemoryOptimized"
  node_size        = "Medium"

  auto_scale {
    min_node_count = 3
    max_node_count = 20
  }

  auto_pause {
    delay_in_minutes = 15
  }
}

resource "azurerm_storage_account" "domain_storage" {
  name                     = "${var.domain_name}datalake"
  resource_group_name      = azurerm_resource_group.rg.name
  location                 = azurerm_resource_group.rg.location
  account_tier             = "Standard"
  account_replication_type = "ZRS"
  is_hns_enabled          = true

  network_rules {
    default_action = "Deny"
    ip_rules       = var.allowed_ips
    bypass         = ["AzureServices"]
  }
}
"""

class DataProductProvisioner:
    """
    Automated provisioning for new data products
    """

    def __init__(self, domain: str, subscription_id: str):
        self.domain = domain
        self.subscription_id = subscription_id

    def provision_data_product(self, product_config: Dict):
        """
        Provision complete infrastructure for a data product
        """

        steps = [
            self._create_storage_container,
            self._setup_spark_pool,
            self._configure_access_policies,
            self._register_in_catalog,
            self._setup_monitoring,
            self._deploy_pipelines
        ]

        for step in steps:
            result = step(product_config)
            if not result["success"]:
                self._rollback(product_config)
                raise Exception(f"Provisioning failed: {result['error']}")

        return {
            "status": "success",
            "product_name": product_config["name"],
            "endpoints": self._get_endpoints(product_config)
        }

    def _create_storage_container(self, config: Dict):
        """Create dedicated storage for data product"""
        container_name = f"{self.domain}-{config['name']}"

        # Create container using Azure SDK
        # Implementation omitted for brevity

        return {"success": True}

    def _setup_spark_pool(self, config: Dict):
        """Configure Spark pool for data product"""
        # Implementation using Azure SDK
        return {"success": True}

    def _configure_access_policies(self, config: Dict):
        """Set up RBAC and access policies"""
        # Implementation using Azure SDK
        return {"success": True}

    def _register_in_catalog(self, config: Dict):
        """Register in Azure Purview"""
        # Implementation using Purview SDK
        return {"success": True}

    def _setup_monitoring(self, config: Dict):
        """Configure Azure Monitor alerts"""
        # Implementation using Monitor SDK
        return {"success": True}

    def _deploy_pipelines(self, config: Dict):
        """Deploy Data Factory pipelines"""
        # Implementation using ADF SDK
        return {"success": True}

from typing import List, Callable

class GovernancePolicy:
    """
    Define and enforce governance policies
    """

    def __init__(self, name: str, scope: str, rules: List[Callable]):
        self.name = name
        self.scope = scope  # "global" or "domain"
        self.rules = rules

    def validate(self, data_product: DataProduct) -> Dict:
        """Validate data product against policy"""
        violations = []

        for rule in self.rules:
            result = rule(data_product)
            if not result["passed"]:
                violations.append(result)

        return {
            "policy_name": self.name,
            "passed": len(violations) == 0,
            "violations": violations
        }

# Define global policies
def gdpr_compliance_check(product: DataProduct) -> Dict:
    """Ensure GDPR compliance"""
    has_pii_classification = "pii" in product.metadata.get("classifications", [])
    has_retention_policy = product.metadata.get("retention_days") is not None

    return {
        "rule": "gdpr_compliance",
        "passed": has_pii_classification and has_retention_policy,
        "message": "GDPR compliance requires PII classification and retention policy"
    }

def encryption_at_rest_check(product: DataProduct) -> Dict:
    """Verify encryption is enabled"""
    is_encrypted = product.metadata.get("encryption_enabled", False)

    return {
        "rule": "encryption_at_rest",
        "passed": is_encrypted,
        "message": "Data must be encrypted at rest"
    }

# Create global policy
global_security_policy = GovernancePolicy(
    name="global_security_policy",
    scope="global",
    rules=[gdpr_compliance_check, encryption_at_rest_check]
)

# Validate data product
validation_result = global_security_policy.validate(customer_product)

if not validation_result["passed"]:
    print(f"Policy violations: {validation_result['violations']}")

# Data product ownership structure
ownership_model = {
    "product_owner": {
        "role": "Business stakeholder",
        "responsibilities": [
            "Define product roadmap",
            "Set quality SLAs",
            "Prioritize features",
            "Approve changes"
        ]
    },
    "technical_lead": {
        "role": "Engineering lead",
        "responsibilities": [
            "Technical implementation",
            "Performance optimization",
            "Infrastructure management",
            "On-call support"
        ]
    },
    "data_steward": {
        "role": "Governance liaison",
        "responsibilities": [
            "Data quality monitoring",
            "Compliance verification",
            "Metadata management",
            "Access control"
        ]
    }
}