Databricks Medallion Architecture

CREATE TABLE IF NOT EXISTS lakehouse.gold.daily_product_performance (
  event_date       DATE,
  product_id       STRING,
  sessions         BIGINT,
  add_to_cart      BIGINT,
  purchases        BIGINT,
  revenue          DECIMAL(18, 2),
  conversion_rate  DOUBLE
)
CLUSTER BY (event_date, product_id);

-- Rebuild only the last 3 days so late-arriving events are absorbed
INSERT INTO lakehouse.gold.daily_product_performance
REPLACE WHERE event_date >= date_sub(current_date(), 3)
SELECT  event_date,
        product_id,
        COUNT(DISTINCT session_id)                                   AS sessions,
        COUNT_IF(event_type = 'add_to_cart')                         AS add_to_cart,
        COUNT_IF(event_type = 'purchase')                            AS purchases,
        CAST(SUM(IF(event_type = 'purchase', amount, 0)) AS DECIMAL(18, 2)) AS revenue,
        COUNT_IF(event_type = 'purchase')
          / NULLIF(COUNT(DISTINCT session_id), 0)                    AS conversion_rate
FROM    lakehouse.silver.events
WHERE   event_date >= date_sub(current_date(), 3)
  AND   product_id IS NOT NULL
GROUP BY event_date, product_id;

OPTIMIZE lakehouse.gold.daily_product_performance;

from delta.tables import DeltaTable
from pyspark.sql import functions as F

CATALOG = "lakehouse"
RAW = "abfss://raw@stlakehouseprod.dfs.core.windows.net/app_events/"
CHK = "abfss://checkpoints@stlakehouseprod.dfs.core.windows.net/app_events"

# ── Bronze: incremental file ingestion with schema evolution
(spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", f"{CHK}/schema")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .load(RAW)
    .withColumn("_source_file", F.col("_metadata.file_path"))
    .withColumn("_ingested_at", F.current_timestamp())
    .writeStream
    .option("checkpointLocation", f"{CHK}/bronze")
    .trigger(availableNow=True)
    .toTable(f"{CATALOG}.bronze.app_events")
    .awaitTermination())


# ── Silver: typed, deduplicated, idempotent upsert
def upsert_events(batch_df, batch_id: int) -> None:
    events = (batch_df
        .where(F.col("event_id").isNotNull())
        .select(
            "event_id", "session_id", "event_type", "product_id",
            F.to_timestamp("event_time").alias("event_ts"),
            F.to_date(F.to_timestamp("event_time")).alias("event_date"),
            F.col("amount").cast("decimal(18,2)").alias("amount"),
        )
        .dropDuplicates(["event_id"]))

    (DeltaTable.forName(spark, f"{CATALOG}.silver.events").alias("t")
        .merge(events.alias("s"), "t.event_id = s.event_id")
        .whenNotMatchedInsertAll()
        .execute())


(spark.readStream.table(f"{CATALOG}.bronze.app_events")
    .writeStream
    .foreachBatch(upsert_events)
    .option("checkpointLocation", f"{CHK}/silver")
    .trigger(availableNow=True)
    .start()
    .awaitTermination())