Transform Data Across Lakehouse Layers

• Bronze to Silver cleaning
• Join operations
• Aggregation and window functions
• Gold layer objects
• DLT expectations

Most pipeline work in Databricks follows a three-layer pattern: Bronze (raw ingested data), Silver (cleaned and conformed data), and Gold (business-ready aggregations and views). Each layer has a specific job to do.

Bronze to Silver cleaning

The Bronze layer stores data exactly as it arrived — no type casting, no deduplication, no filtering. Silver is where cleaning happens.

A typical Bronze-to-Silver transformation handles three things: filtering out bad rows, casting columns to the right types, and removing duplicates.

from pyspark.sql import functions as F
from pyspark.sql.window import Window

bronze = spark.read.table("catalog.bronze.raw_orders")

silver = (
    bronze
    .filter(F.col("order_id").isNotNull() & F.col("customer_id").isNotNull())
    .withColumn("order_ts", F.to_timestamp("order_ts", "yyyy-MM-dd HH:mm:ss"))
    .withColumn("amount", F.col("amount").cast("decimal(18,2)"))
    .withColumn(
        "row_num",
        F.row_number().over(
            Window.partitionBy("order_id").orderBy(F.col("order_ts").desc())
        )
    )
    .filter(F.col("row_num") == 1)
    .drop("row_num")
)

silver.write.format("delta").mode("overwrite").saveAsTable("catalog.silver.orders")

Join operations

Spark supports several join types. The behavior on null keys differs across them:

# inner join — only matching rows on both sides
orders.join(customers, on="customer_id", how="inner")

# left join — all orders, null customer fields if no match
orders.join(customers, on="customer_id", how="left")

# broadcast join — sends the smaller table to each executor
# avoids a shuffle; use when one table fits in memory (< a few hundred MB)
from pyspark.sql.functions import broadcast
orders.join(broadcast(customers), on="customer_id", how="inner")

# cross join — every combination of rows (use with care)
small_table.crossJoin(lookup_table)

Broadcast joins are the most common optimization lever. When the optimizer does not choose a broadcast automatically (controlled by spark.sql.autoBroadcastJoinThreshold, default 10MB), you can force it with broadcast().

Aggregation and window functions

Standard aggregation with groupBy:

summary = (
    silver
    .groupBy("customer_id", F.date_trunc("month", "order_ts").alias("order_month"))
    .agg(
        F.count("order_id").alias("order_count"),
        F.sum("amount").alias("total_spend"),
        F.avg("amount").alias("avg_order_value")
    )
)

Window functions operate on a partition of rows without collapsing them. Useful for rankings, running totals, and prior-row comparisons:

from pyspark.sql.window import Window

win = Window.partitionBy("customer_id").orderBy("order_ts")

enriched = silver.withColumns({
    "order_rank": F.rank().over(win),
    "running_total": F.sum("amount").over(win.rowsBetween(Window.unboundedPreceding, 0)),
    "prev_order_amount": F.lag("amount", 1).over(win)
})

Gold layer objects

The Gold layer holds business-ready data. Databricks supports four object types at this layer, each with different update behavior:

Table: A materialized Delta table. Populated by an explicit write (scheduled job or pipeline). Fast reads since data is pre-computed. Needs a process to keep it current.

View: A saved query, no stored data. Always reflects the underlying tables at query time. No storage cost, but the full query runs on every access.

Materialized View (Delta Live Tables): Databricks computes and stores the result, then updates it incrementally when source data changes. Faster than a plain view, less manual than a scheduled table write.

Streaming Table (Delta Live Tables): Designed for append-only streaming sources. Processes new rows incrementally using a streaming engine.

-- Materialized View in DLT
CREATE OR REFRESH MATERIALIZED VIEW gold.monthly_revenue AS
SELECT
    customer_id,
    date_trunc('month', order_ts) AS order_month,
    SUM(amount) AS total_revenue
FROM silver.orders
GROUP BY 1, 2;

-- Streaming Table in DLT
CREATE OR REFRESH STREAMING TABLE bronze.raw_events AS
SELECT * FROM STREAM(read_files('s3://my-bucket/events/', format => 'json'));

DLT expectations

Delta Live Tables supports data quality rules called expectations. An expectation defines a constraint; what happens when a row fails determines which type to use:

import dlt
from pyspark.sql.functions import col

@dlt.table
@dlt.expect("valid_order_id", "order_id IS NOT NULL")
@dlt.expect_or_drop("positive_amount", "amount > 0")
@dlt.expect_or_fail("required_customer", "customer_id IS NOT NULL")
def silver_orders():
    return spark.read.table("catalog.bronze.raw_orders")

• expect: logs a warning when the constraint fails; the row still flows through
• expect_or_drop: rows that fail are silently dropped from the output
• expect_or_fail: any failing row stops the pipeline with an error

Use expect for informational quality tracking, expect_or_drop when bad rows should be excluded without stopping the pipeline, and expect_or_fail when a failing constraint means the source data cannot be trusted at all.