Data Mesh Architecture with PySpark

Architecture Diagram: Data Mesh Conceptual Model

Architecture Diagram

╔══════════════════════════════════════════════════════════════════════════════════════════════╗
║                          DATA MESH ARCHITECTURE (Conceptual)                                  ║
╠══════════════════════════════════════════════════════════════════════════════════════════════╣
║                                                                                              ║
║  ┌────────────────────────────────────────────────────────────────────────────────────────┐  ║
║  │                    FOUR FOUNDATIONAL PRINCIPLES                                         │  ║
║  │                                                                                        │  ║
║  │  ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────┐  ┌────────────────┐ │  ║
║  │  │   1. DOMAIN      │  │   2. DATA AS     │  │   3. SELF-SERVE  │  │  4. FEDERATED  │ │  ║
║  │  │   ORIENTED       │  │   A PRODUCT      │  │   DATA           │  │  COMPUTATIONAL  │ │  ║
║  │  │                  │  │                  │  │   PLATFORM       │  │  GOVERNANCE     │ │  ║
║  │  │  • Business      │  │  • Discoverable  │  │                  │  │                │ │  ║
║  │  │    domains own   │  │  • Addressable   │  │  • Infrastructure│  │  • Global      │ │  ║
║  │  │    their data    │  │  • Trustworthy   │  │    as Code       │  │    standards   │ │  ║
║  │  │  • Cross-domain  │  │  • Self-describing│ │  • Domain teams  │  │  • Domain      │ │  ║
║  │  │    via products  │  │  • Interoperable │  │    build their   │  │    autonomy    │ │  ║
║  │  │  • No centralized│  │  • Secure        │  │    own pipelines │  │  • Automated   │ │  ║
║  │  │    data team     │  │  • Versioned     │  │  • Platform team │  │    policy      │ │  ║
║  │  │                  │  │                  │  │    provides      │  │    enforcement │ │  ║
║  │  │                  │  │                  │  │    abstractions  │  │                │ │  ║
║  │  └──────────────────┘  └──────────────────┘  └──────────────────┘  └────────────────┘ │  ║
║  └────────────────────────────────────────────────────────────────────────────────────────┘  ║
║                                                                                              ║
║  ┌────────────────────────────────────────────────────────────────────────────────────────┐  ║
║  │                         DOMAIN OWNERSHIP STRUCTURE                                      │  ║
║  │                                                                                        │  ║
║  │                        ┌──────────────────────────────┐                                │  ║
║  │                        │      PLATFORM TEAM            │                                │  ║
║  │                        │  (Infrastructure & Tooling)   │                                │  ║
║  │                        │                              │                                │  ║
║  │                        │  • Data lakehouse infrastructure                              │  ║
║  │                        │  • Compute orchestration (Spark)                              │  ║
║  │                        │  • Identity & access management                               │  ║
║  │                        │  • Observability & monitoring                                  │  ║
║  │                        └──────────────┬───────────────┘                                │  ║
║  │                                       │                                                │  ║
║  │            ┌──────────────────────────┼──────────────────────────┐                     │  ║
║  │            │                          │                          │                      │  ║
║  │            ▼                          ▼                          ▼                      │  ║
║  │  ┌──────────────────┐  ┌──────────────────┐  ┌──────────────────┐                     │  ║
║  │  │  SALES DOMAIN    │  │  MARKETING DOMAIN│  │  SUPPLY CHAIN   │                     │  ║
║  │  │  (Domain Team)   │  │  (Domain Team)   │  │  DOMAIN Team)   │                     │  ║
║  │  │                  │  │                  │  │                  │                     │  ║
║  │  │  ┌────────────┐  │  │  ┌────────────┐  │  │  ┌────────────┐  │                     │  ║
║  │  │  │ Data       │  │  │  │ Data       │  │  │  │ Data       │  │                     │  ║
║  │  │  │ Product:   │  │  │  │ Product:   │  │  │  │ Product:   │  │                     │  ║
║  │  │  │ customer_  │  │  │  │ campaign_  │  │  │  │ inventory_ │  │                     │  ║
║  │  │  │ lifetime_  │  │  │  │ attribution│  │  │  │ forecast   │  │                     │  ║
║  │  │  │ value      │  │  │  │ _model     │  │  │  │            │  │                     │  ║
║  │  │  └────────────┘  │  │  └────────────┘  │  │  └────────────┘  │                     │  ║
║  │  │  ┌────────────┐  │  │  ┌────────────┐  │  │  ┌────────────┐  │                     │  ║
║  │  │  │ Data       │  │  │  │ Data       │  │  │  │ Data       │  │                     │  ║
║  │  │  │ Product:   │  │  │  │ Product:   │  │  │  │ Product:   │  │                     │  ║
║  │  │  │ order_     │  │  │  │ customer_  │  │  │  │ supplier_  │  │                     │  ║
║  │  │  │ analytics  │  │  │  │ segments   │  │  │  │ performance│  │                     │  ║
║  │  │  └────────────┘  │  │  └────────────┘  │  │  └────────────┘  │                     │  ║
║  │  └──────────────────┘  └──────────────────┘  └──────────────────┘                     │  ║
║  └────────────────────────────────────────────────────────────────────────────────────────┘  ║
║                                                                                              ║
╚══════════════════════════════════════════════════════════════════════════════════════════════╝

Architecture Diagram: Data Product Lifecycle

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────────────────┐
│                    DATA PRODUCT LIFECYCLE IN PYSPARK                                 │
├─────────────────────────────────────────────────────────────────────────────────────┤
│                                                                                     │
│  PHASE 1: DESIGN                PHASE 2: BUILD                                     │
│  ┌─────────────────────────┐    ┌─────────────────────────┐                        │
│  │                         │    │                         │                        │
│  │  • Domain Discovery     │    │  • Source Extraction     │                        │
│  │    └─ Identify business │    │    └─ PySpark read from  │                        │
│  │       domain boundaries │    │       domain sources     │                        │
│  │  • Product Charter      │    │  • Data Transformation  │                        │
│  │    └─ Define SLA,       │    │    └─ PySpark transforms │                        │
│  │       freshness, quality│    │       (clean, enrich)    │                        │
│  │  • Schema Design        │    │  • Schema Validation    │                        │
│  │    └─ Define output     │    │    └─ Enforce contracts  │                        │
│  │       schema + types    │    │       via PySpark checks │                        │
│  │  • Access Policy        │    │  • Quality Checks       │                        │
│  │    └─ Define who can    │    │    └─ Automated tests   │                        │
│  │       consume product   │    │       in pipeline       │                        │
│  └────────────┬────────────┘    └────────────┬────────────┘                        │
│               │                              │                                      │
│               ▼                              ▼                                      │
│  PHASE 3: PUBLISH              PHASE 4: OPERATE                                    │
│  ┌─────────────────────────┐    ┌─────────────────────────┐                        │
│  │                         │    │                         │                        │
│  │  • Register in Catalog  │    │  • Monitor Freshness    │                        │
│  │    └─ Unity Catalog /   │    │    └─ Alert if SLA      │                        │
│  │       Data Catalog      │    │       breached           │                        │
│  │  • Version Release      │    │  • Track Usage Metrics  │                        │
│  │    └─ Semantic versioning│   │    └─ Query count,      │                        │
│  │       (v1.0, v1.1)     │    │       consumer count     │                        │
│  │  • Documentation        │    │  • Quality Monitoring   │                        │
│  │    └─ Auto-generated    │    │    └─ Schema drift      │                        │
│  │       data dictionary   │    │       detection          │                        │
│  │  • SLA Commitment       │    │  • Cost Attribution     │                        │
│  │    └─ Published uptime  │    │    └─ Track compute +   │                        │
│  │       and freshness     │    │       storage costs      │                        │
│  └────────────┬────────────┘    └────────────┬────────────┘                        │
│               │                              │                                      │
│               ▼                              ▼                                      │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
│  │                         CONSUMER DISCOVERY & ACCESS                         │   │
│  │                                                                             │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐   │   │
│  │  │  Search &    │  │  Request     │  │  Consume     │  │  Feedback    │   │   │
│  │  │  Discover    │──▶  Access     │──▶  & Analyze   │──▶  & Iterate  │   │   │
│  │  │  (Catalog)   │  │  (API/Policies)│ │  (PySpark)  │  │  (Improve)   │   │   │
│  │  └──────────────┘  └──────────────┘  └──────────────┘  └──────────────┘   │   │
│  │                                                                             │   │
│  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                                                                     │
└─────────────────────────────────────────────────────────────────────────────────────┘

Architecture Diagram: Federated Governance Model

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────────────────┐
│                    FEDERATED COMPUTATIONAL GOVERNANCE                                │
├─────────────────────────────────────────────────────────────────────────────────────┤
│                                                                                     │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
│  │                         GOVERNANCE COUNCIL                                   │   │
│  │  (Cross-domain representatives, meets quarterly)                             │   │
│  │                                                                             │   │
│  │  Responsibilities:                                                          │   │
│  │  • Define global data standards (naming, security, quality)                 │   │
│  │  • Approve cross-domain data product contracts                              │   │
│  │  • Resolve domain disputes over shared data                                 │   │
│  │  • Review and approve platform capabilities                                 │   │
│  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                      │                                              │
│                                      ▼                                              │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
│  │                    GOVERNANCE AS CODE (Automated)                            │   │
│  │                                                                             │   │
│  │  ┌───────────────────────────────────────────────────────────────────────┐  │   │
│  │  │  POLICY ENFORCEMENT (PySpark + Delta Lake)                            │  │   │
│  │  │                                                                       │  │   │
│  │  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐ │  │   │
│  │  │  │  Schema     │  │  Quality    │  │  Security   │  │  SLA        │ │  │   │
│  │  │  │  Registry   │  │  Gates      │  │  Policies   │  │  Monitor    │ │  │   │
│  │  │  │             │  │             │  │             │  │             │ │  │   │
│  │  │  │ • Enforce   │  │ • Null      │  │ • Column    │  │ • Freshness │ │  │   │
│  │  │  │   required  │  │   checks    │  │   masking   │  │   tracking  │ │  │   │
│  │  │  │   columns   │  │ • Range     │  │ • Row-level │  │ • Alert     │ │  │   │
│  │  │  │ • Validate  │  │   validation│  │   filtering │  │   on breach │ │  │   │
│  │  │  │   data types│  │ • Uniqueness│  │ • Encryption│  │ • Auto-     │ │  │   │
│  │  │  │ • Reject    │  │   assertions│  │   at rest   │  │   scaling   │ │  │   │
│  │  │  │   invalid   │  │ • Completeness│ │             │  │             │ │  │   │
│  │  │  │   schemas   │  │   checks    │  │             │  │             │ │  │   │
│  │  │  └─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘ │  │   │
│  │  └───────────────────────────────────────────────────────────────────────┘  │   │
│  │                                                                             │   │
│  │  ┌───────────────────────────────────────────────────────────────────────┐  │   │
│  │  │  IMPLEMENTATION (PySpark Code)                                        │  │   │
│  │  │                                                                       │  │   │
│  │  │  # Schema enforcement                                                │  │   │
│  │  │  enforced_df = validate_schema(df, expected_schema)                   │  │   │
│  │  │                                                                       │  │   │
│  │  │  # Quality gates                                                     │  │   │
│  │  │  quality_report = run_quality_checks(df, quality_rules)               │  │   │
│  │  │  if quality_report.has_critical_failures:                             │  │   │
│  │  │      raise DataQualityException(quality_report)                       │  │   │
│  │  │                                                                       │  │   │
│  │  │  # Security policies                                                 │  │   │
│  │  │  masked_df = apply_column_masking(df, security_policies)              │  │   │
│  │  │                                                                       │  │   │
│  │  │  # SLA monitoring                                                    │  │   │
│  │  │  record_freshness(product_name, current_timestamp())                  │  │   │
│  │  └───────────────────────────────────────────────────────────────────────┘  │   │
│  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                      │                                              │
│                                      ▼                                              │
│  ┌─────────────────────────────────────────────────────────────────────────────┐   │
│  │                    DOMAIN-LEVEL GOVERNANCE                                   │   │
│  │                                                                             │   │
│  │  Each domain team:                                                          │   │
│  │  • Implements global policies within their domain                           │   │
│  │  • Defines domain-specific quality rules                                    │   │
│  │  • Manages access policies for their data products                          │   │
│  │  • Reports compliance to governance council                                 │   │
│  │  • Has autonomy over internal domain data structures                        │   │
│  │                                                                             │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐   │   │
│  │  │  Sales       │  │  Marketing   │  │  Supply Chain│  │  Finance     │   │   │
│  │  │  Domain      │  │  Domain      │  │  Domain      │  │  Domain      │   │   │
│  │  │  Policies    │  │  Policies    │  │  Policies    │  │  Policies    │   │   │
│  │  └──────────────┘  └──────────────┘  └──────────────┘  └──────────────┘   │   │
│  └─────────────────────────────────────────────────────────────────────────────┘   │
│                                                                                     │
└─────────────────────────────────────────────────────────────────────────────────────┘

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.

The four foundational principles of Data Mesh are: (1) Domain-Oriented Ownership — each business domain (sales, marketing, supply chain, finance) owns and manages its data as a first-class product; (2) Data as a Product — domain data is published as discoverable, addressable, trustworthy, self-describing, interoperable, and secure products with published SLAs; (3) Self-Serve Data Platform — a platform team provides infrastructure abstractions (compute, storage, orchestration, cataloging) that enable domain teams to build and operate their own data products without deep platform expertise; (4) Federated Computational Governance — global policies (security, quality, interoperability) are defined by a governance council but enforced programmatically through automated policies embedded in the platform.

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. The platform team provides reusable Spark libraries, shared cluster configurations, and standardized deployment patterns. Domain teams have full autonomy over their transformation logic while adhering to global standards for schema format, quality metrics, and access controls.

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), and documentation. Each data product has a published SLA specifying maximum staleness (e.g., data must be refreshed every 4 hours), minimum quality thresholds (e.g., 99.9% completeness), and availability guarantees.

The federated governance model replaces traditional centralized data governance with automated, code-based policy enforcement. Instead of a governance team manually reviewing data access requests, policies are encoded as PySpark validation functions, Delta Lake constraints, and Unity Catalog ACLs that are automatically enforced when data products are published or consumed.

Key Concepts Table

Mathematical Foundations

Definition: Data Product

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

P = (D, S, Q, A, M)

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:

\text{CDR}(d) = \frac{|\{p \in P_d : \exists d' \neq d, p.\text{depends\_on}(d')\}|}{|P_d|}

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:

\forall d \neq d_{\text{new}}: d.\text{interfaces} \cap d_{\text{new}}.\text{interfaces} = \emptyset

This is the "independent deployability" property.

Federated Governance Cost

Governance overhead scales as:

C_{\text{governance}} = \sum_{i=1}^{n} c_{\text{policy}}(d_i) + n \times c_{\text{coordination}}

where the coordination cost grows linearly with domain count.

Data Product Discovery

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

\text{Score}_{\text{discover}} = \alpha \cdot m_{\text{schema}} + \beta \cdot m_{\text{quality}} + \gamma \cdot m_{\text{lineage}}

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.)

Concept	Description	PySpark Implementation	Ownership
Domain	Business capability boundary	Spark apps per domain	Domain Team
Data Product	Discoverable, trustworthy data unit	Delta tables + metadata	Domain Team
Data Product Contract	Published SLA + schema guarantee	PySpark validation + Unity Catalog	Domain Team
Self-Serve Platform	Infrastructure abstractions	Shared Spark configs, libraries	Platform Team
Federated Governance	Automated policy enforcement	PySpark quality gates, ACLs	Governance Council
Schema Registry	Centralized schema management	Delta Lake schema + Unity Catalog	Platform Team
Lineage Tracking	Data flow documentation	OpenLineage + Spark listeners	Platform Team
Quality Gates	Automated validation	PySpark quality checks in pipeline	Domain Team
Cost Attribution	Per-domain cost tracking	Spark metrics + cloud billing	Platform Team
Cross-Domain Join	Inter-domain data integration	Data product composition	Consumer

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

Metric	Centralized Data Team	Data Mesh (Decentralized)	Improvement
Time to New Data Product	4-8 weeks	1-2 weeks	75% faster
Data Product Count	20-50 (central bottleneck)	200-500 (distributed)	10x scaling
Data Quality Incidents	15/month	5/month	67% reduction
Mean Time to Recovery	4-8 hours	1-2 hours	75% faster
Data Consumer Satisfaction	3.2/5.0	4.5/5.0	41% improvement
Cost per Data Product	$5,000/month \|$ 1,500/month	70% reduction
Cross-Domain Query Latency	30 seconds	5 seconds	83% faster
Governance Policy Compliance	65% (manual)	98% (automated)	51% improvement
Developer Productivity	2 products/engineer	8 products/engineer	4x improvement
Data Freshness (avg)	24 hours	4 hours	83% faster

Best Practices

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.
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.
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.
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.
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.
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.
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.
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.
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.
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: Snowflake Time Travel (snowflake/02), Kafka CDC (kafka/04), Airflow DAGs (airflow/02)

Data Mesh Architecture with PySpark

Data Mesh Architecture with PySpark

Architecture Diagram: Data Mesh Conceptual Model

Architecture Diagram: Data Product Lifecycle

Architecture Diagram: Federated Governance Model

Detailed Explanation

Key Concepts Table

Mathematical Foundations

Definition: Data Product

Domain Ownership Metric

Data Mesh Scalability Theorem

Federated Governance Cost

Data Product Discovery

Key Insight

Summary

Key Concepts Table (cont.)

Code Examples

Example 1: Defining a Data Product with Quality Contracts

Example 2: Cross-Domain Data Product Composition

Example 3: Federated Governance with Automated Quality Gates

Performance Metrics

Best Practices

Need Expert PySpark Help?