Medallion Architecture in Databricks: Vollständiger Implementierungs-Guide

-- Create catalogs per environment
CREATE CATALOG IF NOT EXISTS prod;
CREATE CATALOG IF NOT EXISTS dev;

-- Create schemas for each medallion layer
CREATE SCHEMA IF NOT EXISTS prod.bronze;
CREATE SCHEMA IF NOT EXISTS prod.silver;
CREATE SCHEMA IF NOT EXISTS prod.gold;

from pyspark.sql import SparkSession
from pyspark.sql.functions import current_timestamp, input_file_name, lit

spark = SparkSession.builder.getOrCreate()

def ingest_bronze(source_path: str, target_table: str, schema_hint: str = None):
    # Incrementally ingest raw files into the Bronze layer using Auto Loader

    reader = (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation", f"/mnt/checkpoints/{target_table}/schema")
        .option("cloudFiles.inferColumnTypes", "true")
    )

    if schema_hint:
        reader = reader.schema(schema_hint)

    df = (
        reader.load(source_path)
        .withColumn("_ingested_at", current_timestamp())
        .withColumn("_source_file", input_file_name())
        .withColumn("_batch_id", lit("manual"))
    )

    (
        df.writeStream
        .format("delta")
        .option("checkpointLocation", f"/mnt/checkpoints/{target_table}/stream")
        .option("mergeSchema", "true")
        .outputMode("append")
        .trigger(availableNow=True)
        .table(target_table)
        .awaitTermination()
    )

# Usage
ingest_bronze(
    source_path="abfss://landing@storageaccount.dfs.core.windows.net/events/",
    target_table="prod.bronze.events_raw"
)

from delta.tables import DeltaTable
from pyspark.sql.functions import col, to_timestamp, trim, upper, when, current_timestamp

def transform_silver_events(spark):
    # Clean and standardize raw events into the Silver layer

    bronze_df = spark.read.table("prod.bronze.events_raw")

    silver_df = (
        bronze_df
        .withColumn("event_timestamp", to_timestamp(col("event_ts"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
        .withColumn("user_id", trim(col("user_id")))
        .withColumn("event_type", upper(trim(col("event_type"))))
        .withColumn("amount", col("amount").cast("decimal(18,4)"))
        .filter(col("event_timestamp").isNotNull())
        .filter(col("user_id").isNotNull() & (col("user_id") != ""))
        .dropDuplicates(["event_id", "event_timestamp"])
        .withColumn("_silver_processed_at", current_timestamp())
    )

    return silver_df

def write_silver_merge(df, target_table: str, merge_keys: list):
    # Upsert into Silver using Delta merge for idempotency

    df.limit(0).write.format("delta").mode("ignore").saveAsTable(target_table)

    target = DeltaTable.forName(spark, target_table)

    merge_condition = " AND ".join([f"target.{k} = source.{k}" for k in merge_keys])

    (
        target.alias("target")
        .merge(df.alias("source"), merge_condition)
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute()
    )

silver_df = transform_silver_events(spark)
write_silver_merge(
    df=silver_df,
    target_table="prod.silver.events",
    merge_keys=["event_id"]
)

ALTER TABLE prod.silver.events
  ADD CONSTRAINT chk_amount_positive CHECK (amount >= 0);

ALTER TABLE prod.silver.events
  ADD CONSTRAINT chk_event_type_valid CHECK (event_type IN ('CLICK', 'PURCHASE', 'VIEW', 'SIGN_UP'));

import dlt
from pyspark.sql.functions import col, to_timestamp

@dlt.table(name="events_silver")
@dlt.expect_or_drop("valid_amount", "amount >= 0")
@dlt.expect_or_fail("non_null_user", "user_id IS NOT NULL")
def events_silver():
    return (
        dlt.read("events_raw")
        .withColumn("event_timestamp", to_timestamp(col("event_ts")))
        .dropDuplicates(["event_id"])
    )

from pyspark.sql.functions import (
    col, sum as spark_sum, count, countDistinct,
    date_trunc, avg, current_timestamp
)

def build_gold_daily_revenue(spark):
    # Aggregate Silver events into a daily revenue Gold table

    df = (
        spark.read.table("prod.silver.events")
        .filter(col("event_type") == "PURCHASE")
        .filter(col("event_timestamp") >= "2024-01-01")
    )

    gold_df = (
        df
        .withColumn("event_date", date_trunc("day", col("event_timestamp")))
        .groupBy("event_date", "country", "product_category")
        .agg(
            spark_sum("amount").alias("total_revenue"),
            count("event_id").alias("transaction_count"),
            countDistinct("user_id").alias("unique_buyers"),
            avg("amount").alias("avg_order_value")
        )
        .withColumn("_gold_updated_at", current_timestamp())
    )

    return gold_df

gold_df = build_gold_daily_revenue(spark)

(
    gold_df.write
    .format("delta")
    .mode("overwrite")
    .option("replaceWhere", "event_date >= '2024-01-01'")
    .partitionBy("event_date")
    .saveAsTable("prod.gold.daily_revenue")
)

-- Optimize file sizes for efficient scans
OPTIMIZE prod.gold.daily_revenue ZORDER BY (country, product_category);

-- Enable liquid clustering for automatic optimization (DBR 13.3+)
ALTER TABLE prod.gold.daily_revenue
  CLUSTER BY (event_date, country);

-- Set table properties for BI performance
ALTER TABLE prod.gold.daily_revenue
  SET TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true'
  );

# workflow.yml (Databricks Asset Bundle format)
resources:
  jobs:
    medallion_pipeline:
      name: "Medallion Pipeline - Events"
      tasks:
        - task_key: bronze_ingestion
          notebook_task:
            notebook_path: ./notebooks/bronze_ingestion
          job_cluster_key: bronze_cluster

        - task_key: silver_transform
          depends_on:
            - task_key: bronze_ingestion
          notebook_task:
            notebook_path: ./notebooks/silver_transform
          job_cluster_key: standard_cluster

        - task_key: gold_aggregation
          depends_on:
            - task_key: silver_transform
          notebook_task:
            notebook_path: ./notebooks/gold_aggregation
          job_cluster_key: standard_cluster

from datetime import datetime

def log_layer_metrics(table: str, layer: str, spark):
    # Log row counts and quality metrics for observability
    df = spark.read.table(table)

    metrics = {
        "table": table,
        "layer": layer,
        "row_count": df.count(),
        "timestamp": datetime.now().isoformat()
    }

    print(f"[{layer.upper()}] {table}: {metrics['row_count']:,} rows")
    return metrics