Building the Gold Layer — Materialized Views and Incremental Refresh

Bronze and silver are done — raw CDC events landing untouched, then merged into SCD Type 2 history. This lecture adds the gold layer and completes the pipeline: one materialized view, aggregating silver into something a BI tool or analyst would actually query.

Add a third file to transformations — gold.py — continuing the one-file-per-layer structure.

Writing the Gold Layer

Materialized views are the right destination for the gold layer, and — as covered in the core concepts lecture — MV flow only has an implicit form. @dp.materialized_view is the sole way to declare it.

pythonCopy
from pyspark import pipelines as dp
from pyspark.sql import functions as F

# ─────────────────────────────────────────────────────────────────
# Gold layer — materialized views for business reporting
#
# Materialized views use spark.read (batch semantics). SDP's
# incremental refresh engine processes only new or changed data
# from Silver whenever possible.
#
# Gold reads Silver's CURRENT state — records where __END_AT IS NULL.
# This is the SCD Type 2 current-record filter. Historical rows
# are excluded unless the business requirement is point-in-time.
#
# Never use spark.readStream in a @dp.materialized_view function.
# ─────────────────────────────────────────────────────────────────


@dp.materialized_view(
    name="gold_daily_revenue",
    comment="Daily revenue by customer tier — current orders only (SCD Type 2 open records)"
)
def gold_daily_revenue():
    orders = (
        spark.read.table("silver_orders")
            .filter(F.col("__END_AT").isNull())          # current records only
            .filter(F.col("status") == "completed")      # completed orders only
    )

    customers = (
        spark.read.table("silver_customers")
            .filter(F.col("__END_AT").isNull())          # current records only
            .select("customer_id", "customer_tier")
    )
    return(
        orders
            .join(customers, on="customer_id", how="left")
            .groupBy(
                F.to_date(F.col("order_date")).alias("order_date"),
                F.col("customer_tier")
            )
            .agg(
                F.sum("amount").alias("total_revenue"),
                F.count("order_id").alias("order_count"),
                F.avg("amount").alias("avg_order_value"),
            )
            .orderBy("order_date", "customer_tier")
    )

Why __END_AT IS NULL matters here

This is the payoff for the SCD Type 2 work done in the silver layer. Both silver_orders and silver_customers carry full history — every past version of every record is still there. Gold doesn't want that history; it wants current state only. Filtering __END_AT IS NULL on both tables before joining is exactly what selects "the row that's true right now" and excludes every superseded version. Skip this filter, and the join would double-count customers and orders against their own historical versions.

Why spark.read, never spark.readStream

Bronze and silver are streaming jobs — they process incrementally, appending or merging only what's new. Gold's materialized view flow is fundamentally different: it's a batch job. The code reads the entire silver table every time, with no incremental logic written anywhere. That's not a mistake — it's the correct way to write MV flow code. SDP's incremental refresh engine will still try to refresh this materialized view incrementally behind the scenes when it can; when it can't, it silently falls back to a full recompute. Either way, correctness of the result — not how it got there — is the framework's job, not yours. Using spark.readStream here isn't just unnecessary, it's wrong: gold requires spark.read.

Running the Complete Pipeline

With bronze, silver, and gold all defined, this is the point to run the whole pipeline — not just one file. Since no new files have landed since the silver lecture, bronze and silver re-execute quickly with nothing new to process, and gold computes its aggregation fresh.

Complete pipeline graph — bronze feeding silver feeding gold

This is the full shape of the pipeline: bronze_customers_cdc → silver_customers, bronze_orders_cdc → silver_orders, and both silver tables converging into the single gold_daily_revenue materialized view. All of this dependency ordering — silver waiting for bronze, gold waiting for both — was resolved automatically by SDP just from reading the source files. No job scheduler, no manually declared task order, nothing wired by hand.

gold_daily_revenue comes out with 4 records: one row per (order date, customer tier) combination that showed up in the current, completed orders.

Every one of these tables is now sitting in Unity Catalog exactly like any other managed table — bronze_customers_cdc, bronze_orders_cdc, silver_customers, silver_orders, and gold_daily_revenue. They're queryable by any tool or team with the right catalog permissions, not just from inside the pipeline.

Proving the Pipeline: A Second Batch of CDC Events

The real test of a CDC pipeline isn't the first load — it's what happens when updates and deletes show up. A second batch of files was uploaded to the landing volume:

• cdc_customers_batch2.csv — one update (Alice Johnson) and one delete (Carol White).
• cdc_orders_batch2.csv — two updates, one delete, and one new insert.

Running the full pipeline again picks these up automatically — bronze ingests the new rows exactly as they arrived, silver applies the CDC merge logic, and gold recomputes.

Checking bronze_customers_cdc directly confirms the raw update landed untouched: Alice Johnson now has two rows — the original from batch1, and a new one from batch2 with an updated email and customer tier, marked with op = 'u'. That's bronze doing exactly its job: no interpretation, just faithfully recording that a change happened.

The interesting part is what silver does with it:

silver_customers showing SCD Type 2 history for Alice Johnson and Carol White

Look at __START_AT and __END_AT for the two customers involved:

• Carol White — __END_AT is now populated (no longer NULL), and no new row was opened for her. This is exactly how SCD Type 2 represents a delete: the record isn't physically removed, it's simply closed. Its __END_AT marks the point it stopped being current.
• Alice Johnson — now has two rows. The original (customer_tier = "Gold") has its __END_AT set to the exact same timestamp the new row's __START_AT begins at — the old version was closed at precisely the moment the new one (customer_tier = "Platinum") opened. The new row's __END_AT is NULL, marking it as the current, active version.

This is the entire point of SCD Type 2: nothing is overwritten, and nothing is lost. The full history of how a record changed over time stays queryable, while a simple __END_AT IS NULL filter — exactly what gold's code already does — always resolves to "the current picture."

The orders side behaves the same way: two updates and a delete were correctly merged and closed, and the new order was inserted as a fresh current row. Checking gold_daily_revenue after this run confirms the four aggregated rows correctly reflect all of it.

Idempotency Check

Running the entire pipeline a third time, with no new files added, is worth doing as a final sanity check. Every layer reports zero new records — bronze reads nothing new, silver merges nothing new, and gold's four rows come back unchanged. Whether the materialized view achieved this by an incremental no-op or a full recompute that happened to match, the guarantee holds either way: the result is correct and stable, regardless of how many times the pipeline runs against unchanged source data.

Summary

That completes the end-to-end medallion pipeline — bronze, silver, and gold, built entirely with Spark Declarative Pipelines, five flows total, with every dependency, incremental load, and CDC merge handled by the framework rather than hand-written orchestration code.

See you again. Keep learning, and keep growing!