Databricks Production Lakehouse Architecture: Unity Catalog, Delta Live Tables & Photon

import dlt
from pyspark.sql import functions as F

# 1. Bronze: Raw streaming ingestion with Auto Loader & schema rescue
@dlt.table(
    name="bronze_orders",
    comment="Raw streaming orders ingested via Auto Loader",
    table_properties={"quality": "bronze"}
)
def bronze_orders():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaLocation", "s3://lakehouse-checkpoints/orders_schema")
        .option("cloudFiles.schemaEvolutionMode", "rescue")
        .load("s3://raw-landing-zone/orders/")
    )

# 2. Silver: Cleaned, validated, and deduplicated orders
@dlt.table(
    name="silver_orders",
    comment="Cleaned orders with validated schemas and data quality gates",
    table_properties={"quality": "silver"}
)
@dlt.expect_or_drop("valid_order_id", "order_id IS NOT NULL")
@dlt.expect_or_drop("positive_amount", "amount > 0")
def silver_orders():
    return (
        dlt.read_stream("bronze_orders")
        .filter(F.col("_rescued_data").isNull())
        .withColumn("order_timestamp", F.to_timestamp("order_date"))
        .select("order_id", "customer_id", "amount", "order_timestamp")
        .dropDuplicates(["order_id"])
    )