Medallion Architecture - Lakehouse Pattern¶

# Bronze layer ingestion pattern
from delta.tables import DeltaTable
from pyspark.sql.functions import current_timestamp, input_file_name

# Read raw data from source
raw_df = spark.read \
    .format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("abfss://landing@datalake.dfs.core.windows.net/crm_data/*.csv")

# Add metadata columns
bronze_df = raw_df \
    .withColumn("_ingestion_timestamp", current_timestamp()) \
    .withColumn("_source_file", input_file_name()) \
    .withColumn("_layer", lit("bronze"))

# Write to Bronze layer with Delta Lake
bronze_df.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .partitionBy("_ingestion_date") \
    .save("abfss://bronze@datalake.dfs.core.windows.net/crm/customers")

# Create or update Delta table
DeltaTable.createOrReplace(spark) \
    .location("abfss://bronze@datalake.dfs.core.windows.net/crm/customers") \
    .addColumns(bronze_df.schema) \
    .partitionedBy("_ingestion_date") \
    .property("delta.enableChangeDataFeed", "true") \
    .execute()

# Silver layer transformation pattern
from pyspark.sql.functions import col, when, trim, upper, regexp_replace
from pyspark.sql.types import DateType, IntegerType

# Read from Bronze
bronze_df = spark.read \
    .format("delta") \
    .load("abfss://bronze@datalake.dfs.core.windows.net/crm/customers")

# Data quality transformations
silver_df = bronze_df \
    .filter(col("customer_id").isNotNull()) \
    .withColumn("email", lower(trim(col("email")))) \
    .withColumn("phone", regexp_replace(col("phone"), "[^0-9]", "")) \
    .withColumn("country_code", upper(trim(col("country_code")))) \
    .withColumn("registration_date", col("registration_date").cast(DateType())) \
    .dropDuplicates(["customer_id"]) \
    .withColumn("_silver_processed_timestamp", current_timestamp())

# Data validation
validated_df = silver_df \
    .filter(col("email").rlike("^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$")) \
    .filter(length(col("phone")) >= 10) \
    .filter(col("country_code").isin(["US", "CA", "UK", "DE", "FR"]))

# Implement SCD Type 2 for history tracking
from delta.tables import DeltaTable

silver_delta = DeltaTable.forPath(spark,
    "abfss://silver@datalake.dfs.core.windows.net/crm/dim_customers")

# Merge with SCD Type 2 logic
silver_delta.alias("target").merge(
    validated_df.alias("source"),
    "target.customer_id = source.customer_id AND target.is_current = true"
).whenMatchedUpdate(
    condition = """
        target.email != source.email OR
        target.phone != source.phone OR
        target.country_code != source.country_code
    """,
    set = {
        "is_current": "false",
        "end_date": "current_date()",
        "updated_timestamp": "current_timestamp()"
    }
).whenNotMatchedInsert(
    values = {
        "customer_id": "source.customer_id",
        "email": "source.email",
        "phone": "source.phone",
        "country_code": "source.country_code",
        "start_date": "current_date()",
        "end_date": "to_date('9999-12-31')",
        "is_current": "true",
        "created_timestamp": "current_timestamp()"
    }
).execute()

# Gold layer aggregation pattern
from pyspark.sql.functions import sum, count, avg, max, min, last_day

# Read from Silver
dim_customers = spark.read.format("delta") \
    .load("abfss://silver@datalake.dfs.core.windows.net/crm/dim_customers")

fact_orders = spark.read.format("delta") \
    .load("abfss://silver@datalake.dfs.core.windows.net/sales/fact_orders")

# Create business-level aggregation
customer_metrics = fact_orders \
    .join(dim_customers, "customer_id") \
    .groupBy(
        "customer_id",
        "country_code",
        last_day(col("order_date")).alias("month_ending")
    ) \
    .agg(
        count("order_id").alias("total_orders"),
        sum("order_amount").alias("total_revenue"),
        avg("order_amount").alias("avg_order_value"),
        max("order_date").alias("last_order_date"),
        min("order_date").alias("first_order_date")
    ) \
    .withColumn("customer_lifetime_months",
        months_between(col("last_order_date"), col("first_order_date")))

# Write to Gold layer with optimization
customer_metrics.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("country_code", "month_ending") \
    .save("abfss://gold@datalake.dfs.core.windows.net/analytics/customer_metrics")

# Optimize with Z-ordering for common query patterns
from delta.tables import DeltaTable

gold_delta = DeltaTable.forPath(spark,
    "abfss://gold@datalake.dfs.core.windows.net/analytics/customer_metrics")

gold_delta.optimize().executeZOrderBy("customer_id", "month_ending")

from pyspark.sql.functions import col, when, count, isnan, isnull

class DataQualityChecker:
    """Data Quality framework for Medallion Architecture"""

    def __init__(self, spark, layer):
        self.spark = spark
        self.layer = layer
        self.quality_metrics = []

    def check_completeness(self, df, required_columns):
        """Check for NULL values in required columns"""
        for col_name in required_columns:
            null_count = df.filter(
                col(col_name).isNull() | isnan(col(col_name))
            ).count()

            total_count = df.count()
            completeness_pct = ((total_count - null_count) / total_count) * 100

            self.quality_metrics.append({
                "layer": self.layer,
                "check_type": "completeness",
                "column": col_name,
                "null_count": null_count,
                "total_count": total_count,
                "completeness_pct": completeness_pct,
                "passed": null_count == 0
            })

        return self

    def check_uniqueness(self, df, unique_columns):
        """Check for duplicate records"""
        total_count = df.count()
        distinct_count = df.select(unique_columns).distinct().count()
        duplicate_count = total_count - distinct_count

        self.quality_metrics.append({
            "layer": self.layer,
            "check_type": "uniqueness",
            "column": ",".join(unique_columns),
            "duplicate_count": duplicate_count,
            "total_count": total_count,
            "uniqueness_pct": (distinct_count / total_count) * 100,
            "passed": duplicate_count == 0
        })

        return self

    def check_validity(self, df, column, validation_expr):
        """Check data validity against business rules"""
        invalid_count = df.filter(~validation_expr).count()
        total_count = df.count()

        self.quality_metrics.append({
            "layer": self.layer,
            "check_type": "validity",
            "column": column,
            "invalid_count": invalid_count,
            "total_count": total_count,
            "validity_pct": ((total_count - invalid_count) / total_count) * 100,
            "passed": invalid_count == 0
        })

        return self

    def get_quality_report(self):
        """Generate quality report DataFrame"""
        from pyspark.sql import Row

        report_df = self.spark.createDataFrame(
            [Row(**metric) for metric in self.quality_metrics]
        )

        return report_df

    def save_quality_metrics(self, path):
        """Save quality metrics to Delta table"""
        report_df = self.get_quality_report()

        report_df.write \
            .format("delta") \
            .mode("append") \
            .partitionBy("layer") \
            .save(path)

# Example usage
silver_df = spark.read.format("delta") \
    .load("abfss://silver@datalake.dfs.core.windows.net/crm/customers")

quality_checker = DataQualityChecker(spark, "silver")

quality_checker \
    .check_completeness(silver_df, ["customer_id", "email", "country_code"]) \
    .check_uniqueness(silver_df, ["customer_id"]) \
    .check_validity(silver_df, "email",
        col("email").rlike("^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$"))

quality_report = quality_checker.get_quality_report()
quality_report.show()

quality_checker.save_quality_metrics(
    "abfss://gold@datalake.dfs.core.windows.net/monitoring/data_quality_metrics"
)

# Optimize file sizes during write
df.coalesce(10).write \
    .format("delta") \
    .mode("append") \
    .option("maxRecordsPerFile", 1000000) \
    .save("path/to/delta/table")

# Post-write optimization
from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "path/to/delta/table")
delta_table.optimize().executeCompaction()

# Appropriate partitioning by layer
# Bronze - partition by ingestion date
bronze_df.write \
    .format("delta") \
    .partitionBy("_ingestion_date") \
    .save("bronze/table")

# Silver - partition by business dimension
silver_df.write \
    .format("delta") \
    .partitionBy("country_code", "year", "month") \
    .save("silver/table")

# Gold - partition for optimal query performance
gold_df.write \
    .format("delta") \
    .partitionBy("region", "year_month") \
    .save("gold/table")

from delta.tables import DeltaTable

# Z-order frequently queried columns
gold_table = DeltaTable.forPath(spark,
    "abfss://gold@datalake.dfs.core.windows.net/analytics/sales_summary")

gold_table.optimize().executeZOrderBy("customer_id", "product_category", "date")

# Standard metadata columns
metadata_columns = [
    "_ingestion_timestamp",      # When data was ingested
    "_source_system",            # Source system identifier
    "_source_file",              # Original source file
    "_processing_timestamp",     # When transformation occurred
    "_layer",                    # Current layer (bronze/silver/gold)
    "_data_quality_score",       # Quality score (0-100)
    "_is_current",               # For SCD Type 2
    "_effective_start_date",     # For SCD Type 2
    "_effective_end_date"        # For SCD Type 2
]

from pyspark.sql.functions import expr

# Quarantine bad records in Bronze
try:
    clean_df = bronze_df.filter(col("customer_id").isNotNull())
    quarantine_df = bronze_df.filter(col("customer_id").isNull())

    # Write clean records
    clean_df.write.format("delta").mode("append") \
        .save("bronze/customers/clean")

    # Write quarantined records for investigation
    quarantine_df.write.format("delta").mode("append") \
        .partitionBy("_ingestion_date") \
        .save("bronze/customers/quarantine")

except Exception as e:
    # Log error and save to error table
    error_df = spark.createDataFrame([{
        "error_timestamp": current_timestamp(),
        "table_name": "customers",
        "layer": "bronze",
        "error_message": str(e),
        "record_count": bronze_df.count()
    }])

    error_df.write.format("delta").mode("append") \
        .save("monitoring/processing_errors")

    raise

from delta.tables import DeltaTable
from pyspark.sql.functions import max as spark_max

def incremental_bronze_to_silver(source_path, target_path, watermark_column):
    """Incremental processing from Bronze to Silver"""

    # Get last processed watermark
    try:
        silver_table = DeltaTable.forPath(spark, target_path)
        last_watermark = silver_table.toDF() \
            .select(spark_max(watermark_column)) \
            .collect()[0][0]
    except:
        last_watermark = "1900-01-01"

    # Read only new data from Bronze
    bronze_df = spark.read.format("delta").load(source_path) \
        .filter(col(watermark_column) > last_watermark)

    # Apply Silver transformations
    silver_df = bronze_df \
        .dropDuplicates(["customer_id"]) \
        .filter(col("customer_id").isNotNull())

    # Merge into Silver
    silver_table.alias("target").merge(
        silver_df.alias("source"),
        "target.customer_id = source.customer_id"
    ).whenMatchedUpdateAll() \
     .whenNotMatchedInsertAll() \
     .execute()

# Usage
incremental_bronze_to_silver(
    "bronze/crm/customers",
    "silver/crm/dim_customers",
    "_ingestion_timestamp"
)