Data Mesh Architecture with PySpark

Data Mesh Architecture

Architecture Diagram: Data Mesh Conceptual Model

Architecture Diagram: Data Product Lifecycle

Architecture Diagram: Federated Governance Model

Detailed Explanation

Data Mesh is a decentralized sociotechnical approach to data architecture that treats data as a product owned by domain teams rather than a centralized data engineering team. Founded by Zhamak Dehghani, Data Mesh addresses the scaling limitations of centralized data architectures (data warehouses, monolithic data lakes) by distributing ownership, computation, and governance across autonomous business domains.

What are the Four Foundational Principles of Data Mesh?

How is Data Mesh Implemented in PySpark?

In PySpark implementations, each domain team maintains its own set of Spark applications that extract data from operational sources, transform it into well-defined data products, and publish them to a shared data lakehouse.

Domain Team Responsibilities:

• Full autonomy over transformation logic
• Adherence to global standards for schema format, quality metrics, and access controls

Platform Team Responsibilities:

• Provide reusable Spark libraries
• Shared cluster configurations
• Standardized deployment patterns

What are Data Products in Data Mesh?

Data products in Data Mesh are more than just tables — they are self-contained units of data with:

• Metadata: Schema, description, lineage
• Quality metrics: Freshness, completeness, accuracy
• Access policies: Who can read, who can write
• Documentation: Comprehensive product documentation
• Published SLAs: Maximum staleness (e.g., data must be refreshed every 4 hours), minimum quality thresholds (e.g., 99.9% completeness), and availability guarantees

What is Federated Computational Governance?

The federated governance model replaces traditional centralized data governance with automated, code-based policy enforcement.

How it Works:

• Policies are encoded as PySpark validation functions, Delta Lake constraints, and Unity Catalog ACLs
• Automatically enforced when data products are published or consumed
• No manual review of data access requests by a governance team

Key Takeaway: Data Mesh shifts from centralized data teams to domain-oriented ownership, treating data as a product. The trade-off is increased coordination overhead but reduced single points of failure.

Key Concepts Table

Mathematical Foundations

Definition: Data Product

A data product $P$ is a self-contained, discoverable data asset defined as:

where $D$ is the dataset, $S$ is the schema, $Q$ is quality SLAs, $A$ is access controls, and $M$ is metadata. Data products must be addressable, trustworthy, and self-describing.

Domain Ownership Metric

Domain independence is measured by the cross-domain dependency ratio:

Target: $\text{CDR}(d) < 0.2$ for healthy domain boundaries.

Data Mesh Scalability Theorem

Adding a new domain $d_{\text{new}}$ increases system capacity by $\Delta C = C(d_{\text{new}})$ without modifying existing domains if:

This is the "independent deployability" property.

Federated Governance Cost

Governance overhead scales as:

where the coordination cost grows linearly with domain count.

Data Product Discovery

Findability score for a data product with metadata completeness $m$:

where $\alpha + \beta + \gamma = 1$ are weighting factors.

Key Insight

Data mesh shifts from centralized data teams to domain-oriented ownership. The trade-off is increased coordination overhead but reduced single points of failure. Platform teams provide self-service infrastructure to minimize domain onboarding cost.

Summary

Data mesh distributes data ownership to domains, each managing data as products. Cross-domain dependency ratio measures boundary health. Governance overhead scales linearly with domains. The architecture trades coordination complexity for organizational scalability.

Key Concepts Table (cont.)

Code Examples

Example 1: Defining a Data Product with Quality Contracts

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
import json

spark = SparkSession.builder \
    .appName("Data-Product-Sales") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

# ─── Data Product Definition ───
@dataclass
class DataProductContract:
    """Defines the contract for a data product."""
    name: str
    version: str
    domain: str
    owner: str
    description: str
    schema: StructType
    sla_freshness_hours: int = 4
    quality_rules: Dict = field(default_factory=dict)
    access_policy: Dict = field(default_factory=dict)
    dependencies: List[str] = field(default_factory=list)

# Define the Sales Domain Data Product
customer_ltv_product = DataProductContract(
    name="customer_lifetime_value",
    version="1.2.0",
    domain="sales",
    owner="sales-data-team@company.com",
    description="Customer lifetime value calculation based on historical orders",
    schema=StructType([
        StructField("customer_id", StringType(), False),
        StructField("customer_name", StringType(), True),
        StructField("ltv_score", DoubleType(), True),
        StructField("total_revenue", DecimalType(18, 2), True),
        StructField("order_count", IntegerType(), True),
        StructField("avg_order_value", DecimalType(18, 2), True),
        StructField("first_order_date", DateType(), True),
        StructField("last_order_date", DateType(), True),
        StructField("segment", StringType(), True),
        StructField("_computed_at", TimestampType(), False),
        StructField("_product_version", StringType(), False),
    ]),
    sla_freshness_hours=4,
    quality_rules={
        "customer_id": {"not_null": True, "unique": True},
        "ltv_score": {"not_null": True, "min": 0, "max": 1000000},
        "total_revenue": {"not_null": True, "min": 0},
        "order_count": {"not_null": True, "min": 1},
        "completeness_threshold": 0.99,
    },
    access_policy={
        "owner_domain": "sales",
        "allowed_consumers": ["marketing", "finance", "executive"],
        "pii_columns": ["customer_name"],
        "masking_rules": {"customer_name": "partial_mask"},
    },
    dependencies=["orders_raw", "customers_master"]
)

# ─── Build the Data Product ───
def build_customer_ltv(spark, contract):
    """Build the customer lifetime value data product."""
    
    # Read source data from domain-owned Bronze layer
    orders = spark.read.format("delta").load("/mnt/sales/bronze/orders")
    customers = spark.read.format("delta").load("/mnt/sales/bronze/customers")
    
    # Transform: Calculate LTV metrics
    ltv_df = (
        orders
        .join(customers, "customer_id", "left")
        .groupBy(
            orders["customer_id"],
            customers["customer_name"]
        )
        .agg(
            count("*").alias("order_count"),
            sum("amount").alias("total_revenue"),
            avg("amount").alias("avg_order_value"),
            min("order_date").alias("first_order_date"),
            max("order_date").alias("last_order_date"),
        )
        .withColumn("ltv_score",
            col("total_revenue") * 0.3 +
            col("order_count") * 10 +
            (current_date().datediff(col("last_order_date")) * -0.1)
        )
        .withColumn("segment",
            when(col("ltv_score") > 10000, "platinum")
            .when(col("ltv_score") > 5000, "gold")
            .when(col("ltv_score") > 1000, "silver")
            .otherwise("bronze")
        )
        .withColumn("_computed_at", current_timestamp())
        .withColumn("_product_version", lit(contract.version))
    )
    
    return ltv_df

# Build the product
customer_ltv = build_customer_ltv(spark, customer_ltv_product)

# ─── Quality Validation ───
def validate_data_product(df, contract):
    """Validate data product against its contract."""
    validation_results = []
    
    for col_name, rules in contract.quality_rules.items():
        if col_name == "completeness_threshold":
            continue
        
        col_obj = df[col_name]
        
        if rules.get("not_null"):
            null_count = df.filter(col_obj.isNull()).count()
            total_count = df.count()
            completeness = 1 - (null_count / total_count) if total_count > 0 else 0
            threshold = contract.quality_rules.get("completeness_threshold", 0.99)
            
            validation_results.append({
                "column": col_name,
                "check": "not_null",
                "passed": completeness >= threshold,
                "completeness": completeness,
                "threshold": threshold,
            })
        
        if rules.get("unique"):
            duplicate_count = (
                df.groupBy(col_name)
                .count()
                .filter(col("count") > 1)
                .count()
            )
            validation_results.append({
                "column": col_name,
                "check": "unique",
                "passed": duplicate_count == 0,
                "duplicate_count": duplicate_count,
            })
        
        if "min" in rules:
            min_val = df.agg(min(col_obj)).first()[0]
            validation_results.append({
                "column": col_name,
                "check": "min_value",
                "passed": min_val >= rules["min"],
                "actual_min": min_val,
                "expected_min": rules["min"],
            })
        
        if "max" in rules:
            max_val = df.agg(max(col_obj)).first()[0]
            validation_results.append({
                "column": col_name,
                "check": "max_value",
                "passed": max_val <= rules["max"],
                "actual_max": max_val,
                "expected_max": rules["max"],
            })
    
    return validation_results

# Run validation
results = validate_data_product(customer_ltv, customer_ltv_product)
for r in results:
    status = "PASS" if r["passed"] else "FAIL"
    print(f"[{status}] {r['column']}.{r['check']}: {r}")

# ─── Publish Data Product ───
(
    customer_ltv
    .write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .save("/mnt/sales/gold/customer_lifetime_value")
)

# Register in catalog (Unity Catalog example)
spark.sql("""
    CREATE TABLE IF NOT EXISTS sales.customer_lifetime_value
    USING DELTA
    LOCATION '/mnt/sales/gold/customer_lifetime_value'
    COMMENT 'Customer lifetime value calculation - Sales Domain Product v1.2.0'
    TBLPROPERTIES (
        'product.owner' = 'sales-data-team@company.com',
        'product.version' = '1.2.0',
        'product.sla.freshness.hours' = '4',
        'product.domain' = 'sales',
        'product.quality.completeness' = '0.99'
    )
""")

Example 2: Cross-Domain Data Product Composition

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder \
    .appName("Cross-Domain-Composition") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

# ─── Compose Data Products from Multiple Domains ───
# Marketing domain wants a unified customer view combining:
# - Sales domain: customer_lifetime_value
# - Marketing domain: customer_segments
# - Supply chain domain: delivery_preferences

# Read data products from each domain (via catalog or direct path)
sales_ltv = (
    spark.read
    .format("delta")
    .load("/mnt/sales/gold/customer_lifetime_value")
    .select(
        "customer_id",
        col("ltv_score").alias("sales_ltv_score"),
        col("segment").alias("sales_segment"),
        "total_revenue",
        "order_count",
    )
)

marketing_segments = (
    spark.read
    .format("delta")
    .load("/mnt/marketing/gold/customer_segments")
    .select(
        "customer_id",
        col("segment").alias("marketing_segment"),
        col("engagement_score"),
        col("preferred_channel"),
        col("last_campaign_response"),
    )
)

supply_preferences = (
    spark.read
    .format("delta")
    .load("/mnt/supply_chain/gold/delivery_preferences")
    .select(
        "customer_id",
        col("preferred_delivery_speed"),
        col("avg_delivery_satisfaction"),
        col("return_rate"),
    )
)

# Compose: Join all data products into a unified view
unified_customer = (
    sales_ltv
    .join(marketing_segments, "customer_id", "left")
    .join(supply_preferences, "customer_id", "left")
    .withColumn("unified_score",
        (col("sales_ltv_score") * 0.4) +
        (col("engagement_score") * 0.3) +
        ((1 - col("return_rate")) * 100 * 0.3)
    )
    .withColumn("recommended_action",
        when(
            (col("sales_ltv_score") > 5000) &
            (col("engagement_score") > 70) &
            (col("avg_delivery_satisfaction") > 4.0),
            "premium_retention_offer"
        ).when(
            (col("sales_ltv_score") < 1000) &
            (col("engagement_score") < 30),
            "win_back_campaign"
        ).otherwise("standard_nurture")
    )
    .withColumn("_composed_at", current_timestamp())
    .withColumn("_source_products", array(
        lit("sales.customer_lifetime_value"),
        lit("marketing.customer_segments"),
        lit("supply_chain.delivery_preferences")
    ))
)

# Write composed data product
(
    unified_customer
    .write
    .format("delta")
    .mode("overwrite")
    .save("/mnt/marketing/gold/unified_customer_view")
)

# Register with lineage metadata
spark.sql("""
    CREATE TABLE IF NOT EXISTS marketing.unified_customer_view
    USING DELTA
    LOCATION '/mnt/marketing/gold/unified_customer_view'
    COMMENT 'Unified customer view composed from sales, marketing, and supply chain domains'
    TBLPROPERTIES (
        'product.owner' = 'marketing-data-team@company.com',
        'product.version' = '1.0.0',
        'product.domain' = 'marketing',
        'product.composed_from' = 'sales.customer_lifetime_value,marketing.customer_segments,supply_chain.delivery_preferences',
        'product.sla.freshness.hours' = '8'
    )
""")

Example 3: Federated Governance with Automated Quality Gates

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
import json

spark = SparkSession.builder \
    .appName("Federated-Governance") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

# ─── Global Governance Policies (Defined by Council) ───
GOVERNANCE_POLICIES = {
    "naming_conventions": {
        "table_names": "snake_case",
        "column_names": "snake_case",
        "prefix_required": True,  # e.g., dim_, fact_, agg_
    },
    "security": {
        "pii_detection": True,
        "encryption_at_rest": True,
        "column_masking_required": True,
        "row_level_security": True,
    },
    "quality": {
        "minimum_completeness": 0.99,
        "freshness_sla_hours": 24,
        "uniqueness_required": True,
        "schema_evolution_requires_approval": True,
    },
    "interoperability": {
        "standard_date_columns": ["created_at", "updated_at"],
        "standard_id_columns": ["_id", "id"],
        "timestamp_timezone": "UTC",
        "decimal_precision": {"precision": 18, "scale": 2},
    }
}

# ─── Automated Quality Gate Function ───
class GovernanceGate:
    """Automated governance validation for data products."""
    
    def __init__(self, policies):
        self.policies = policies
        self.violations = []
    
    def validate_schema(self, df, product_name):
        """Validate naming conventions."""
        columns = df.columns
        
        for col_name in columns:
            # Check snake_case
            if not all(c.islower() or c == '_' or c.isdigit() for c in col_name):
                self.violations.append({
                    "product": product_name,
                    "rule": "naming_conventions",
                    "severity": "WARNING",
                    "message": f"Column '{col_name}' not in snake_case"
                })
        
        return len(self.violations) == 0
    
    def validate_quality(self, df, product_name):
        """Validate quality thresholds."""
        total_rows = df.count()
        if total_rows == 0:
            self.violations.append({
                "product": product_name,
                "rule": "quality",
                "severity": "CRITICAL",
                "message": "Data product is empty"
            })
            return False
        
        for col_name in df.columns:
            null_count = df.filter(col(col_name).isNull()).count()
            completeness = 1 - (null_count / total_rows)
            
            if completeness < self.policies["quality"]["minimum_completeness"]:
                self.violations.append({
                    "product": product_name,
                    "rule": "quality.completeness",
                    "severity": "CRITICAL",
                    "column": col_name,
                    "completeness": completeness,
                    "threshold": self.policies["quality"]["minimum_completeness"]
                })
        
        return True
    
    def validate_security(self, df, product_name, pii_columns):
        """Validate security policies."""
        for col_name in pii_columns:
            if col_name not in df.columns:
                self.violations.append({
                    "product": product_name,
                    "rule": "security.pii",
                    "severity": "CRITICAL",
                    "message": f"PII column '{col_name}' not found for masking"
                })
        
        return True
    
    def validate_interoperability(self, df, product_name):
        """Validate interoperability standards."""
        # Check timestamp columns are in UTC
        for col_name in df.columns:
            if col_name.endswith("_at") or col_name.endswith("_timestamp"):
                col_type = dict(df.dtypes).get(col_name)
                if col_type not in ("timestamp", "date"):
                    self.violations.append({
                        "product": product_name,
                        "rule": "interoperability.timestamp",
                        "severity": "WARNING",
                        "column": col_name,
                        "actual_type": col_type,
                        "expected_type": "timestamp"
                    })
        
        return True
    
    def get_report(self):
        """Generate governance validation report."""
        critical = [v for v in self.violations if v["severity"] == "CRITICAL"]
        warnings = [v for v in self.violations if v["severity"] == "WARNING"]
        
        return {
            "total_violations": len(self.violations),
            "critical": len(critical),
            "warnings": len(warnings),
            "passed": len(critical) == 0,
            "violations": self.violations
        }

# ─── Apply Governance to a Data Product ───
gate = GovernanceGate(GOVERNANCE_POLICIES)

# Build a data product
customer_data = spark.read.format("delta").load("/mnt/sales/bronze/customers")
customer_product = (
    customer_data
    .withColumn("ltv_score", col("total_revenue") * 0.3)
    .withColumn("_computed_at", current_timestamp())
)

# Run governance checks
gate.validate_schema(customer_product, "sales.customer_lifetime_value")
gate.validate_quality(customer_product, "sales.customer_lifetime_value")
gate.validate_security(customer_product, "sales.customer_lifetime_value",
                       pii_columns=["customer_name", "email"])
gate.validate_interoperability(customer_product, "sales.customer_lifetime_value")

# Get report
report = gate.get_report()
print(json.dumps(report, indent=2, default=str))

# Block publication if critical violations
if not report["passed"]:
    raise Exception(
        f"Data product 'sales.customer_lifetime_value' failed governance validation. "
        f"Critical violations: {report['critical']}. "
        f"Fix issues before publishing."
    )

Performance Metrics

Best Practices

1. Start with clear domain boundaries — Use Event Storming or Domain-Driven Design to identify bounded contexts before assigning data ownership. Each domain should have clear business capabilities and minimal coupling to other domains.

2. Define data product contracts upfront — Every data product must have a published contract specifying schema, SLA (freshness, availability), quality thresholds, and access policies. Use PySpark validation functions to enforce these contracts programmatically.

3. Build a self-serve platform first — Invest in platform abstractions (shared Spark configs, cluster templates, deployment pipelines, catalog integration) before scaling domain teams. The platform should reduce the cognitive load for domain teams.

4. Implement federated governance as code — Replace manual governance processes with automated PySpark validation functions, Delta Lake constraints, and Unity Catalog policies. Governance should be enforced at publish time, not reviewed manually.

5. Enable discoverability through a data catalog — Use Unity Catalog, DataHub, or Amundsen to register data products with metadata, lineage, and quality metrics. Domain teams should be able to discover and request access to other domains' products.

6. Track data product usage and value — Monitor query patterns, consumer count, and business impact for each data product. Use this data to prioritize improvements and justify domain team investments.

7. Implement cross-domain composition carefully — When composing data products from multiple domains, document dependencies, validate that all source products meet their SLAs, and handle partial failures gracefully.

8. Maintain backward compatibility — Version data products using semantic versioning (v1.0, v1.1, v2.0). Breaking changes require a new major version and deprecation period for existing consumers.

9. Invest in data observability — Deploy monitoring for schema drift, data quality degradation, freshness violations, and cost anomalies. Alert domain teams proactively rather than waiting for consumers to report issues.

10. Foster a data product culture — Train domain teams on data engineering best practices, provide templates and examples, and create communities of practice across domains for knowledge sharing.

See also: Production Hardening (40), Cost Optimization (39), Data Lakehouse Architecture (34)

See Also

• Data Mesh Architecture — Data mesh design principles
• Data Governance — Data governance and cataloging
• Data Contracts — Data contract patterns
• Production Hardening — Production best practices