Build a Medallion Pipeline in Databricks

Build a bronze-silver-gold data pipeline with Delta Lake and Unity Catalog.

Introduction

30 Second Summary

Every online store collects thousands of orders a day, but that raw data is messy. Customer IDs go missing, quantities turn negative, and orders get dated in the future.

In this project, you will build a data pipeline in Databricks using the medallion architecture pattern. Raw e-commerce orders flow through three layers of transformation until you can query clean, ranked product revenue with a single command.

What You'll Build

Run one query and instantly see your top-selling products ranked by revenue, built from raw order data that was full of errors just minutes earlier.

By the end of this project, you'll have:

  • A three-layer medallion pipeline where raw e-commerce orders flow through bronze, silver, and gold Delta Lake tables into ranked product revenue you can query instantly.
  • A MERGE-based incremental upsert that handles new and updated orders without creating duplicates.
  • A time travel recovery that rolls back a corrupted data load, proving your pipeline bounces back from bad writes.
  • An optional Secret Mission to push your skills further.

Are there any prerequisites?

No prerequisites needed. Databricks Free Edition is perpetually free and runs entirely in your browser, with no cloud account or local installations required.

Before We Start

Before you dive into any setup, take a moment to lock in what you are about to build and why it matters.

Set Up Your Databricks Workspace

Before you can build any data pipeline, you need a workspace to build it in. Your pipeline will move data through three layers (bronze, silver, gold), and each layer needs its own dedicated space in a data catalog.

In this step, you will sign up for Databricks Free Edition and create the catalog structure that organizes your medallion layers.

In this step, get ready to:
  • Sign up for Databricks Free Edition and open your workspace.
  • Create a Python notebook connected to serverless compute.
  • Build a Unity Catalog structure with a catalog, three schemas, and a managed volume.
Sign up for Databricks Free Edition

Databricks Free Edition is a perpetually free version of Databricks. It runs on serverless compute, requires no cloud account, and includes Unity Catalog out of the box.

  • Open your browser and go to databricks.com/try-databricks.
  • Select the Free Edition option.
  • Fill in your name, email, and create a password to complete registration.
  • Once your workspace is provisioned, click into it to open the Databricks workspace UI.

What is Databricks Free Edition?

Databricks Free Edition replaced the older Community Edition (retired January 1, 2026). It gives you one workspace and one metastore per account, with daily usage limits but no billing.

Your notebooks automatically connect to serverless compute, so there is nothing to configure on the infrastructure side.

Stuck on the signup page?

Make sure you select Free Edition specifically. The page may also offer paid trial options for other Databricks tiers.

If your workspace takes more than a few minutes to provision, try refreshing the page. Help me sign up for Databricks Free Edition.

Create a notebook and build your catalog structure

Databricks notebooks are where you write and run code. Your notebook needs to use Python as its language, and it will automatically connect to serverless compute.

  • In the left sidebar of your workspace, click + New.
  • Select Notebook.
  • Confirm the notebook language is set to Python in the language dropdown at the top of the notebook.
  • Confirm the notebook shows a compute connection in the top-right area of the notebook. On Free Edition, this defaults to serverless compute.

Now you need to create the catalog structure that organizes your medallion layers. Unity Catalog uses a three-level namespace: catalog.schema.table.

What is Unity Catalog?

Unity Catalog is Databricks' governance layer for all your data assets. It organizes everything into catalogs, schemas, and tables (or volumes).

You will create one catalog called medallion_project, then a schema inside it for each medallion layer: bronze, silver, and gold. A managed volume inside bronze will serve as the landing zone for raw data files.

  • In the first cell of your notebook, paste the following code to create the catalog, schemas, and volume:
# Create the catalog to hold all medallion layers
spark.sql("CREATE CATALOG IF NOT EXISTS medallion_project")

# Create one schema per medallion layer
spark.sql("CREATE SCHEMA IF NOT EXISTS medallion_project.bronze")
spark.sql("CREATE SCHEMA IF NOT EXISTS medallion_project.silver")
spark.sql("CREATE SCHEMA IF NOT EXISTS medallion_project.gold")

# Create a managed volume for landing raw data files
spark.sql("CREATE VOLUME IF NOT EXISTS medallion_project.bronze.raw_data")

# Verify the schemas exist
spark.sql("SHOW SCHEMAS IN medallion_project")

What does this code do?

  • CREATE CATALOG IF NOT EXISTS medallion_project creates a top-level container that holds all of your project's data objects.
  • The three CREATE SCHEMA statements create one schema per medallion layer. Tables you create later will live inside these schemas.
  • CREATE VOLUME creates a managed volume inside the bronze schema. Volumes are storage locations for raw files before they are loaded into tables.
  • SHOW SCHEMAS IN medallion_project lists all schemas inside the catalog so you can confirm the structure was created.

Before you run this cell, what schemas do you expect to see listed in the output?

  • Click the Run button (or press Shift+Enter) to execute the cell.

You should see a table in the cell output listing bronze, silver, and gold as schema names. You may also see default and information_schema listed. Those are created automatically by Unity Catalog.

Not seeing the three schemas?

  • Make sure your notebook is connected to compute. Look for a green connected indicator near the top of the notebook.
  • If you see a permissions error, confirm you are using the workspace owner account you signed up with.
  • Check that the catalog name medallion_project is spelled exactly as shown. Catalog and schema names are case-sensitive.

My notebook cell isn't producing the expected schema list. Help me debug.

✔️ Awesome, I've got everything!

Great. Double check that you have saved your notebook.

ⓧ I'd like to double check the full code

Here is the complete code for Cell 1. Compare it with what you have in your notebook:

spark.sql("CREATE CATALOG IF NOT EXISTS medallion_project")
spark.sql("CREATE SCHEMA IF NOT EXISTS medallion_project.bronze")
spark.sql("CREATE SCHEMA IF NOT EXISTS medallion_project.silver")
spark.sql("CREATE SCHEMA IF NOT EXISTS medallion_project.gold")
spark.sql("CREATE VOLUME IF NOT EXISTS medallion_project.bronze.raw_data")
spark.sql("SHOW SCHEMAS IN medallion_project")

Your workspace is ready and your catalog structure is in place. Next up, you will generate raw e-commerce order data and ingest it into the bronze layer.

Ingest Raw Data into the Bronze Layer

Your Databricks workspace is ready with a catalog, three schemas, and a managed volume. Now you need data to flow through the pipeline.

The bronze layer captures raw data exactly as it arrives, with no transformations. This is your safety net. If anything goes wrong downstream, you can always rebuild from bronze.

In this step, get ready to:
  • Generate 500 rows of synthetic e-commerce order data with intentional quality issues.
  • Ingest the raw data into a bronze Delta Lake table.
  • Confirm that dirty records landed alongside clean ones.
Generate synthetic order data

Real pipelines ingest data from source systems. For this project, you will generate synthetic e-commerce orders using PySpark. The data will contain intentional quality issues that the silver layer will catch later.

  • Create a new notebook cell by clicking the + Code button below your existing Cell 1.
  • Set up the imports, schema definition, and product catalog by pasting this code into the new cell:
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, DateType
from pyspark.sql.functions import col, lit, expr, when, rand, concat
from datetime import date, timedelta
import random

# Define schema
order_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DoubleType(), True),
    StructField("order_date", DateType(), True)
])

# Generate 500 rows of synthetic order data
random.seed(42)
products = ["PROD-001", "PROD-002", "PROD-003", "PROD-004", "PROD-005",
            "PROD-006", "PROD-007", "PROD-008", "PROD-009", "PROD-010"]
prices = {"PROD-001": 29.99, "PROD-002": 49.99, "PROD-003": 9.99,
           "PROD-004": 149.99, "PROD-005": 19.99, "PROD-006": 79.99,
           "PROD-007": 39.99, "PROD-008": 59.99, "PROD-009": 14.99,
           "PROD-010": 99.99}

What does this code do?

  • order_schema defines the exact shape of each order row. Notice customer_id is True for nullable, because some records will intentionally have missing customers.
  • random.seed(42) locks the random number generator so every run produces the same dataset. This makes the pipeline reproducible.
  • products and prices define ten products with fixed unit prices, simulating a product catalog.
  • Add the data generation loop, the write to bronze, and the verification queries below the code you just pasted in the same cell:
data = []
for i in range(500):
    order_id = f"ORD-{i+1:05d}"
    # Introduce ~5% null customer_ids
    customer_id = None if random.random() < 0.05 else f"CUST-{random.randint(1, 200):04d}"
    product_id = random.choice(products)
    # Introduce ~3% negative quantities
    quantity = -random.randint(1, 5) if random.random() < 0.03 else random.randint(1, 10)
    unit_price = prices[product_id]
    # Introduce ~2% future dates
    if random.random() < 0.02:
        order_date = date.today() + timedelta(days=random.randint(30, 365))
    else:
        order_date = date.today() - timedelta(days=random.randint(0, 90))

    data.append((order_id, customer_id, product_id, quantity, unit_price, order_date))

# Create DataFrame and write to bronze
bronze_df = spark.createDataFrame(data, schema=order_schema)
bronze_df.writeTo("medallion_project.bronze.orders_raw").createOrReplace()

# Verify
print("Total bronze rows:")
display(spark.sql("SELECT count(*) as total_rows FROM medallion_project.bronze.orders_raw"))
print("Bad records in bronze:")
display(spark.sql("""
    SELECT * FROM medallion_project.bronze.orders_raw
    WHERE quantity < 0 OR customer_id IS NULL
"""))

What does this code do?

  • The loop generates 500 order rows, each with a unique order_id. About 5% get a None customer ID, 3% get a negative quantity, and 2% get a future date. These are the dirty records that the silver layer will catch.
  • writeTo("medallion_project.bronze.orders_raw").createOrReplace() writes the DataFrame as a managed Delta table in your bronze schema. If the table already exists, it gets replaced.
  • The two display() calls at the end query the table you just created. The first shows the total row count. The second shows only the bad records.

Before you run this cell, how many total rows do you expect to see? Will the bad records (nulls, negative quantities, future dates) be included in the count, or filtered out?

  • Run the cell by clicking the play button at the top left of the cell, or pressing Shift+Enter.

You should see a table showing total_rows as 500. All 500 rows landed, including the dirty ones. That is bronze working as designed.

Not seeing 500 rows?

  • Make sure you ran Cell 1 (the setup cell) first. The catalog and schemas must exist before you can write a table to them.
  • Check that both code blocks are in the same cell, in order. The second block references order_schema and products from the first block.
  • If you see a connection error, confirm your notebook is still attached to serverless compute by checking the dropdown at the top right of the notebook.

Help me debug why my bronze table has no rows.

Confirm the dirty data landed in bronze

Scroll down to the second output table from the cell you just ran. This table shows every row where quantity < 0 or customer_id IS NULL.

You should see a handful of rows with null values in the customer_id column and rows with negative numbers in the quantity column.

Why does bronze keep bad data?

Bronze is a raw, unfiltered copy of the source. Cleaning and filtering happen in the silver layer, not here.

This separation matters because it preserves the original data as evidence. If your cleaning logic has a bug, you can fix it and re-process from bronze without going back to the source system.

✔️ Awesome, I've got everything!

Great. Double check that your cell shows 500 total rows in the first output table and a handful of bad records in the second.

ⓧ I'd like to double check the full code

Your Cell 2 should contain two sections of code, in this order:

  • Imports, schema definition, and product catalog (the first block you pasted, starting with from pyspark.sql.types import ... and ending with the prices dictionary).
  • Data generation loop, write, and verify (the second block you pasted, starting with data = [] and ending with the display() calls).

Scroll up to each chunk and confirm they are both present in the same cell, in order.

Your bronze table is loaded with 500 rows of raw order data, dirty records and all. Next up, you will clean and validate this data in the silver layer, separating the good records from the bad.

Clean and Validate Data in the Silver Layer

Your bronze table has 500 rows of raw order data sitting in it. Some of those rows have missing customer IDs, negative quantities, or order dates set in the future.

If you aggregate directly from bronze, those bad records corrupt your revenue totals and product rankings. The silver layer is where you enforce data quality rules. Clean records go to one table, and bad records get quarantined in another so nothing slips through to your gold analytics.

In this step, get ready to:
  • Filter clean records from bronze and quarantine rejected rows with labeled rejection reasons.
  • Write the clean and quarantine DataFrames to separate silver Delta Lake tables.
  • Add a NOT NULL constraint to enforce schema rules on the silver table.
Filter clean records and quarantine bad data

The silver layer applies three data quality rules to every record from bronze. Records must have a non-null customer_id, a positive quantity, and an order_date that is not in the future.

  • Create a new cell below Cell 2 in your notebook. Start building Cell 3 by copying in the imports, the bronze table read, and the clean data filter:
from pyspark.sql.functions import current_date, col, lit, when
from datetime import date

# Read bronze
bronze = spark.read.table("medallion_project.bronze.orders_raw")

# Define quality rules
today = date.today()
clean_df = bronze.filter(
    (col("customer_id").isNotNull()) &
    (col("quantity") > 0) &
    (col("order_date") <= lit(today))
)

What does this code do?

spark.read.table() reads the full bronze table into a DataFrame. This is the starting point for all silver-layer transformations.

The .filter() call chains three conditions with & (logical AND). Only rows that pass all three rules end up in clean_df.

Records that fail any quality rule should not be silently deleted. They go to a quarantine table with a label explaining why they were rejected.

  • Below the clean filter code in the same cell, add the quarantine logic, the write operations for both tables, and the verification queries:
# Quarantine bad records with rejection reasons
quarantine_df = bronze.filter(
    (col("customer_id").isNull()) |
    (col("quantity") <= 0) |
    (col("order_date") > lit(today))
).withColumn(
    "rejection_reason",
    when(col("customer_id").isNull(), "NULL customer_id")
    .when(col("quantity") <= 0, "Non-positive quantity")
    .when(col("order_date") > lit(today), "Future order date")
    .otherwise("Unknown")
)

# Write clean records to silver
clean_df.writeTo("medallion_project.silver.orders_clean").createOrReplace()

# Write quarantined records
quarantine_df.writeTo("medallion_project.silver.orders_quarantine").createOrReplace()

# Verify
print("Clean silver rows:")
display(spark.sql("SELECT count(*) as clean_count FROM medallion_project.silver.orders_clean"))
print("Quarantined rows:")
display(spark.sql("SELECT count(*) as quarantine_count FROM medallion_project.silver.orders_quarantine"))
display(spark.sql("SELECT rejection_reason, count(*) as cnt FROM medallion_project.silver.orders_quarantine GROUP BY rejection_reason"))

What does this code do?

  • The quarantine filter is the inverse of the clean filter. It uses | (logical OR) to catch any record that fails at least one rule.
  • .withColumn("rejection_reason", when(...)) adds a new column that labels each bad record with the specific reason it was rejected. This gives you an audit trail.
  • writeTo().createOrReplace() writes each DataFrame to its own managed Delta table in the silver schema.

Before you run this cell, think about what you expect to see. You started with 500 rows in bronze. Will the clean count be higher or lower than 500? How many quarantined rows do you expect?

  • Run Cell 3 by clicking the play button or pressing Shift+Enter.

The clean count is lower than 500. The quarantine count shows the rejected records, broken down by rejection reason: NULL customer_id, Non-positive quantity, and Future order date.

Why keep quarantined records instead of deleting them?

Silently dropping bad records makes debugging impossible. When a dashboard number looks wrong, the quarantine table lets you trace exactly which records were excluded and why. In production pipelines, this audit trail is critical for data governance.

Seeing an error when running Cell 3?

  • If you see a table not found error for orders_raw, go back to Cell 2 and re-run it. The bronze table must exist before this cell can read from it.
  • If you see a syntax error on the .filter() call, make sure each condition is wrapped in parentheses and the & / | operators connect them.

Help me debug an error in my silver layer filtering code.

Add a NOT NULL constraint to the silver table

Your clean filter already excludes null customer_id values, but that rule only exists in your Python code. If someone writes directly to the silver table and skips the filter, nulls could sneak in. A NOT NULL constraint enforces the rule at the table level.

  • Add the NOT NULL constraint by appending this line to the bottom of Cell 3:
# Add NOT NULL constraint
spark.sql("ALTER TABLE medallion_project.silver.orders_clean ALTER COLUMN customer_id SET NOT NULL")
print("NOT NULL constraint added to customer_id")

What does this constraint do?

ALTER TABLE ... ALTER COLUMN ... SET NOT NULL tells Delta Lake to reject any future write that tries to insert a null value into that column. This works because your clean filter already removed all nulls. The constraint makes the rule permanent and enforced by the storage engine itself.

  • Re-run Cell 3 by clicking the play button or pressing Shift+Enter. You should see NOT NULL constraint added to customer_id printed at the bottom of the output.

Now test that the constraint actually works. What do you think will happen if you try to insert a row with a null customer_id?

  • Create a new cell below Cell 3 and test the constraint by running this SQL insert:
# Test NOT NULL constraint - this should fail
spark.sql("""
    INSERT INTO medallion_project.silver.orders_clean
    VALUES ('TEST-NULL', NULL, 'PROD-001', 1, 29.99, current_date())
""")

The cell throws an error because the NOT NULL constraint on customer_id rejected the insert. That is exactly what you want. The table is now protecting itself.

Clean up your test cell

You can delete the test cell now. It served its purpose. Right-click the cell and select the delete option, or click the cell menu (the three dots at the top right of the cell) and choose Delete.

Constraint error says existing data has nulls?

The NOT NULL constraint can only be applied if every existing row in the table already has a non-null value in that column. If you see this error, your clean filter may not have removed all null customer_id rows. Double check that the .filter() condition includes col("customer_id").isNotNull(), then re-run Cell 3 from the top.

Help me fix a NOT NULL constraint error on my silver table.

Verify your silver layer end to end

Before you move on, confirm the full picture. How many rows do you expect in orders_clean versus orders_quarantine? The two counts should add up to your original 500 bronze rows.

  • Scroll up to Cell 3's output and confirm the clean count is fewer than 500 and the quarantine count accounts for the difference.
  • Confirm the quarantine breakdown shows counts for NULL customer_id, Non-positive quantity, and Future order date.

✔️ Awesome, I've got everything!

Double check that Cell 3 is saved. Your silver layer is ready.

ⓧ I'd like to double check the full code

Here is the complete Cell 3 code. Make sure your cell matches this exactly:

from pyspark.sql.functions import current_date, col, lit, when
from datetime import date

bronze = spark.read.table("medallion_project.bronze.orders_raw")

today = date.today()
clean_df = bronze.filter(
    (col("customer_id").isNotNull()) &
    (col("quantity") > 0) &
    (col("order_date") <= lit(today))
)

quarantine_df = bronze.filter(
    (col("customer_id").isNull()) |
    (col("quantity") <= 0) |
    (col("order_date") > lit(today))
).withColumn(
    "rejection_reason",
    when(col("customer_id").isNull(), "NULL customer_id")
    .when(col("quantity") <= 0, "Non-positive quantity")
    .when(col("order_date") > lit(today), "Future order date")
    .otherwise("Unknown")
)

clean_df.writeTo("medallion_project.silver.orders_clean").createOrReplace()
quarantine_df.writeTo("medallion_project.silver.orders_quarantine").createOrReplace()

spark.sql("ALTER TABLE medallion_project.silver.orders_clean ALTER COLUMN customer_id SET NOT NULL")

Your silver layer is locked down. Clean records are separated from bad data, every rejected record has a reason, and the table itself enforces a NOT NULL rule on customer_id. Next up, you will aggregate this clean data into a gold table that answers a real business question: which products generate the most revenue?

Aggregate into the Gold Layer and Break It

Your silver table has clean, validated order data. The final layer of the medallion architecture is gold, where you pre-compute the business metrics that stakeholders actually query.

In this step, you'll aggregate product revenue rankings into a gold table. Then you'll discover firsthand what happens when bad data slips past silver and poisons the analytics.

In this step, get ready to:
  • Aggregate silver data into a gold table with product revenue rankings.
  • Deliberately corrupt the pipeline with garbage data to see how it breaks gold analytics.
Aggregate product revenue into the gold table

The gold layer answers a specific business question. In this case: "What are the top-selling products by revenue?" Instead of re-running a complex query every time, the gold table stores the pre-computed answer.

  • Create a new cell in your notebook by clicking the + button below Cell 3.
  • Build the gold aggregation by pasting this code into the new cell:
from pyspark.sql.functions import sum as _sum, count, avg, col

# Read from silver
silver = spark.read.table("medallion_project.silver.orders_clean")

# Aggregate by product
gold_df = silver.groupBy("product_id").agg(
    _sum(col("quantity") * col("unit_price")).alias("total_revenue"),
    count("order_id").alias("total_orders"),
    avg(col("quantity") * col("unit_price")).alias("avg_order_value")
)

# Write to gold
gold_df.writeTo("medallion_project.gold.product_revenue").createOrReplace()

# Verify - top 5 products by revenue
print("Top 5 products by revenue:")
display(spark.sql("""
    SELECT * FROM medallion_project.gold.product_revenue
    ORDER BY total_revenue DESC
    LIMIT 5
"""))

What does this code do?

  • groupBy("product_id") groups all orders by product, so each product gets one summary row.
  • _sum(col("quantity") * col("unit_price")) calculates total revenue per product by multiplying quantity by price for every order, then summing.
  • writeTo().createOrReplace() writes the result to a managed Delta Lake table in the gold schema.

Before you run this cell, what do you expect to see? Think about how many distinct products exist in the dataset.

  • Run the cell by clicking the Run button (or pressing Shift+Enter).

You should see a table showing the top 5 products ranked by total_revenue. Each row shows a product_id like PROD-004 alongside its revenue, order count, and average order value.

Not seeing any results?

  • Make sure Cell 3 ran successfully first. The gold aggregation reads from medallion_project.silver.orders_clean, which Cell 3 creates.
  • Check that the import line reads sum as _sum (with the underscore). PySpark aliases sum to avoid conflicting with Python's built-in.
  • help me debug my gold aggregation query

Why pre-aggregate in gold?

Without a gold table, every dashboard query would scan the entire silver table and re-compute the aggregations from scratch. The gold layer runs that expensive calculation once and stores the result.

Downstream consumers (dashboards, reports, analysts) query gold directly. Fast reads, no wasted compute.

Deliberately corrupt the pipeline

Your gold table looks great right now. But in production, data quality problems are not a question of "if", they are a question of "when". What happens if 50 rows of garbage data sneak into your silver table?

Before you run the next cell, predict: if a product called CORRUPT appeared in silver with a quantity of 99999 and unit price of 999.99, where would it land in the gold revenue rankings?

  • Create a new cell below Cell 4 by clicking the + button.
  • Inject corrupt data into silver and re-aggregate gold by pasting this code:
# --- CORRUPT THE PIPELINE ---
corrupt_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("customer_id", StringType(), False),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DoubleType(), True),
    StructField("order_date", DateType(), True)
])
corrupt_data = [(f"CORRUPT-{i}", "CUST-BAD", "CORRUPT", 99999, 999.99, date.today()) for i in range(50)]
corrupt_df = spark.createDataFrame(corrupt_data, schema=corrupt_schema)
corrupt_df.write.mode("append").saveAsTable("medallion_project.silver.orders_clean")

# Re-aggregate gold with corrupted data
silver_corrupted = spark.read.table("medallion_project.silver.orders_clean")
gold_corrupted = silver_corrupted.groupBy("product_id").agg(
    _sum(col("quantity") * col("unit_price")).alias("total_revenue"),
    count("order_id").alias("total_orders"),
    avg(col("quantity") * col("unit_price")).alias("avg_order_value")
)
gold_corrupted.writeTo("medallion_project.gold.product_revenue").createOrReplace()

print("CORRUPTED gold table - notice 'CORRUPT' at the top:")
display(spark.sql("""
    SELECT * FROM medallion_project.gold.product_revenue
    ORDER BY total_revenue DESC
    LIMIT 5
"""))

What does this code do?

  • It creates 50 rows of garbage data with product_id = "CORRUPT", a quantity of 99999, and a price of 999.99.
  • write.mode("append") adds those rows to the existing silver table without replacing it. This simulates a bad data load landing alongside your clean records.
  • The second half re-runs the exact same gold aggregation from Cell 4. Since the silver table now contains corrupt rows, the aggregation blindly includes them.
  • Run the cell by pressing Shift+Enter.

Look at the results. A product called CORRUPT now sits at the top of the revenue rankings with a massively inflated total. The real products that were leading before have been pushed down.

This is exactly the problem. A single bad data load poisoned the entire gold layer, and any dashboard reading from this table would now show completely wrong numbers.

Why is this by design?

You just experienced the most common data engineering failure mode: garbage in, garbage out. Blindly re-running aggregations when the source data is compromised silently corrupts everything downstream.

This is why Delta Lake tracks every change as a versioned snapshot. In the next step, you will use time travel to roll the silver table back to before the corruption happened.

Not seeing CORRUPT in the results?

  • Make sure Cell 4 ran before this cell. The gold aggregation imports (_sum, count, avg) must already be available from running Cell 4.
  • Confirm the StructType and StructField types are still imported from Cell 2. If you restarted the kernel, re-run Cells 1 through 4 first.
  • help me debug the pipeline corruption step

✔️ Awesome, I've got everything!

Great. You should have two new cells in your notebook: Cell 4 for the gold aggregation and Cell 5 for the corruption. Make sure both are saved.

ⓧ I'd like to double check the full code

Since the code for this step spans two cells, scroll up and confirm each chunk is present in order:

  • Cell 4 starts with from pyspark.sql.functions import sum as _sum, count, avg, col and ends with the display() call showing the top 5 products.
  • Cell 5 starts with the # --- CORRUPT THE PIPELINE --- comment, creates 50 corrupt rows, appends them to silver, re-aggregates gold, and ends with a display() showing the corrupted rankings.

Your gold table is poisoned. A fake product is sitting at the top of the revenue rankings, and any stakeholder querying this table right now would see completely wrong data. In the next step, you'll use Delta Lake's time travel to roll back to the clean version and build an incremental upsert pipeline that prevents duplicates.

Recover with Time Travel and Build Incremental Upserts

In the previous step, you saw what happens when bad data slips into your pipeline. A single append of 50 corrupt rows turned your gold revenue rankings upside down, putting a fake product at the top.

Delta Lake stores every version of your table as a separate snapshot. You can roll back to any previous version using time travel, and then use MERGE-based upserts to handle incremental updates without creating duplicates.

In this step, get ready to:
  • Restore the silver table using Delta Lake time travel.
  • Build a MERGE-based incremental upsert.
  • Verify the full pipeline history with DESCRIBE HISTORY.
Inspect version history and restore the silver table

Every write to a Delta table is recorded as a numbered version. Before you can fix anything, you need to see which versions exist and identify the clean one.

  • Create a new cell in your notebook.
  • View the silver table's version history by running this code:
# View the version history of the silver table
display(spark.sql("DESCRIBE HISTORY medallion_project.silver.orders_clean"))

What should I see?

You should see two versions listed. Version 0 is the original createOrReplace from when you wrote clean data in Step 3. Version 1 is the append that injected the 50 corrupt rows in Step 4.

Version 0 is your target. That is the snapshot you want to roll back to.

Now that you know version 0 holds the clean data, you can restore it and rebuild the gold table.

  • Create another new cell in your notebook. This will be your Cell 5.
  • Restore the silver table and rebuild the gold aggregation by pasting the following code, then running the cell:
from delta.tables import DeltaTable
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, DateType
from pyspark.sql.functions import col, sum as _sum, count, avg
from datetime import date, timedelta
import random

# Restore silver to the version before corruption
spark.sql("RESTORE TABLE medallion_project.silver.orders_clean TO VERSION AS OF 0")

# Re-aggregate gold from restored silver
silver_restored = spark.read.table("medallion_project.silver.orders_clean")
gold_restored = silver_restored.groupBy("product_id").agg(
    _sum(col("quantity") * col("unit_price")).alias("total_revenue"),
    count("order_id").alias("total_orders"),
    avg(col("quantity") * col("unit_price")).alias("avg_order_value")
)
gold_restored.writeTo("medallion_project.gold.product_revenue").createOrReplace()

What does this code do?

  • RESTORE TABLE ... TO VERSION AS OF 0 rolls the silver table back to the snapshot before the corrupt append. The corrupt rows are removed.
  • The gold re-aggregation reads from the now-clean silver table and rewrites product_revenue with accurate numbers.

Before you check the result, predict: do you think the CORRUPT product will still appear in the gold table?

  • Verify the recovery by running this query in a new cell:
# Verify CORRUPT is gone from the gold table
display(spark.sql("""
    SELECT * FROM medallion_project.gold.product_revenue
    ORDER BY total_revenue DESC
    LIMIT 5
"""))

The CORRUPT product is gone. Your top 5 products should all be legitimate PROD- entries with realistic revenue figures.

Still seeing CORRUPT in the results?

  • Make sure you ran Cell 5 before the verification query. The restore and re-aggregation must execute first.
  • Confirm the RESTORE command uses VERSION AS OF 0 (not version 1, which is the corrupt version).

Why is my RESTORE not removing the corrupt data?

Build an incremental upsert with MERGE

Your pipeline can now recover from bad data. The next challenge is handling incremental updates. When new orders arrive or existing orders get modified, you need to update the silver table without creating duplicate rows.

What is a MERGE upsert?

MERGE compares incoming rows against existing rows using a key (like order_id). If a match is found, the existing row is updated. If no match is found, the new row is inserted.

This "update or insert" pattern is called an upsert. It is the standard way production pipelines handle incremental data loads.

You will create a batch of 20 orders: 10 updates to existing order IDs (with new quantities) and 10 brand new orders.

  • In Cell 5, add the following code below the gold re-aggregation you already have. This generates the batch data:
# MERGE upsert
random.seed(99)
products = ["PROD-001", "PROD-002", "PROD-003", "PROD-004", "PROD-005"]
prices = {"PROD-001": 29.99, "PROD-002": 49.99, "PROD-003": 9.99,
           "PROD-004": 149.99, "PROD-005": 19.99}

update_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("customer_id", StringType(), False),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DoubleType(), True),
    StructField("order_date", DateType(), True)
])

batch_data = []
# 10 updates to existing orders (changing quantities)
for i in range(1, 11):
    order_id = f"ORD-{i:05d}"
    product = random.choice(products)
    batch_data.append((order_id, f"CUST-{random.randint(1,200):04d}", product,
                        random.randint(5, 20), prices[product], date.today()))
# 10 new orders
for i in range(501, 511):
    order_id = f"ORD-{i:05d}"
    product = random.choice(products)
    batch_data.append((order_id, f"CUST-{random.randint(1,200):04d}", product,
                        random.randint(1, 10), prices[product], date.today()))

batch_df = spark.createDataFrame(batch_data, schema=update_schema)

What does this code do?

  • The first loop generates updates to orders ORD-00001 through ORD-00010. These order IDs already exist in the silver table, so the MERGE will update them.
  • The second loop generates new orders ORD-00501 through ORD-00510. These IDs do not exist yet, so the MERGE will insert them.
  • Still in Cell 5, add the MERGE operation and final gold re-aggregation below the batch data code:
# Execute the MERGE upsert into silver
delta_table = DeltaTable.forName(spark, "medallion_project.silver.orders_clean")
(delta_table.alias("target")
    .merge(batch_df.alias("source"), "target.order_id = source.order_id")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute())

# Re-aggregate gold with upserted data
silver_final = spark.read.table("medallion_project.silver.orders_clean")
gold_final = silver_final.groupBy("product_id").agg(
    _sum(col("quantity") * col("unit_price")).alias("total_revenue"),
    count("order_id").alias("total_orders"),
    avg(col("quantity") * col("unit_price")).alias("avg_order_value")
)
gold_final.writeTo("medallion_project.gold.product_revenue").createOrReplace()

What does the MERGE do?

  • DeltaTable.forName(spark, "medallion_project.silver.orders_clean") opens the silver table as a Delta target.
  • The .merge() call joins the source batch against the target on order_id.
  • .whenMatchedUpdateAll() replaces all columns for matching rows. .whenNotMatchedInsertAll() adds rows with no match as new records.
  • Run Cell 5 to execute the full restoration and upsert pipeline.

Seeing an error about NOT NULL or constraint violations?

  • The RESTORE operation resets table properties including constraints. If you see a constraint-related error on the MERGE, re-run the cell. The restore should clear the conflicting state.
  • Make sure customer_id values in your batch data are not None. The code generates valid customer IDs for every row.

Help me debug my Delta Lake MERGE upsert error.

Verify the upsert and inspect pipeline history

The MERGE should have updated 10 existing orders and inserted 10 new ones. Before you check, predict: will the silver table now have more rows than it originally did, or the same number?

  • Verify the updated and new orders by running this query in a new cell:
# Check specific orders that were updated or inserted
display(spark.sql("""
    SELECT * FROM medallion_project.silver.orders_clean
    WHERE order_id IN ('ORD-00001', 'ORD-00005', 'ORD-00501', 'ORD-00510')
    ORDER BY order_id
"""))

You should see four rows. ORD-00001 and ORD-00005 are existing orders with updated quantities (between 5 and 20). ORD-00501 and ORD-00510 are brand new orders that did not exist before.

  • Verify the gold table reflects the upserted data by running:
# Verify gold table reflects updated revenue
display(spark.sql("""
    SELECT * FROM medallion_project.gold.product_revenue
    ORDER BY total_revenue DESC
"""))

Your gold table should show all 10 products ranked by revenue with no CORRUPT entries. The revenue totals now include the upserted quantities.

  • Run DESCRIBE HISTORY one more time to see the full timeline of operations:
# View the complete version history
display(spark.sql("DESCRIBE HISTORY medallion_project.silver.orders_clean"))

What should the version history show?

You should see four versions, each telling the story of what happened to this table:

  • Version 0: the original clean data write from Step 3.
  • Version 1: the corrupt append from Step 4.
  • Version 2: the RESTORE you just ran, rolling back to version 0.
  • Version 3: the MERGE upsert that applied your 20-row batch.

This is a complete audit trail. Every operation is logged, and any version can be restored.

Seeing more versions than expected?

  • If you ran Cell 5 more than once, each execution adds extra RESTORE and MERGE entries. The data is still correct, but the version numbers will be higher.
  • Check that the most recent operation is a MERGE. If the latest entry is a RESTORE, Cell 5 may have errored partway through. Re-run it.

Help me understand my Delta table version history.

✔️ Awesome, I've got everything!

Great. Double check that your Cell 5 is saved. It should contain all three sections listed below.

ⓧ I'd like to double check the full code

Your Cell 5 should contain all of the following sections in order. Scroll through it and confirm each one is present:

  • Imports: DeltaTable, PySpark types and functions, date, timedelta, and random.
  • Restore + gold re-aggregation: RESTORE TABLE ... TO VERSION AS OF 0, then reading from silver and writing aggregated results to medallion_project.gold.product_revenue.
  • Batch data generation: random.seed(99), 5-product subset, schema definition, two loops generating 10 updates (ORD-00001 to ORD-00010) and 10 new orders (ORD-00501 to ORD-00510).
  • MERGE upsert + final gold re-aggregation: DeltaTable.forName() with .merge().whenMatchedUpdateAll().whenNotMatchedInsertAll().execute(), then reading from silver and writing final aggregation to gold.

Your pipeline can now ingest raw data, clean it, aggregate it, recover from corruption, and handle incremental updates. That is a production-ready medallion architecture. Next up, take on the Secret Mission to add automated data quality constraints and freshness monitoring.

Secret mission

Add Data Quality Constraints and Freshness Monitoring

Your pipeline works, but it relies on manual data quality checks that run only when you remember to run them. In production, bad data can slip in at 3 AM when nobody is watching. Can you add guardrails that fire automatically and a monitor that alerts you when the pipeline goes stale?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on Databricks Free Edition, so there are no ongoing costs. Your workspace and data persist across sessions at no charge.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Unity Catalog catalog medallion_project with bronze, silver, and gold schemas.
  • Managed volume medallion_project.bronze.raw_data.
  • Delta tables: orders_raw, orders_clean, orders_quarantine, and product_revenue.
  • Python notebook with all pipeline cells.

Keep everything running

No action needed. Choose this if you plan to come back and extend your pipeline or experiment with the data.

  • Your Databricks Free Edition workspace stays active at no cost.
  • All Delta tables, schemas, and the medallion_project catalog persist across sessions.
  • Your notebook with all pipeline cells remains in the workspace.

Pause - I'll come back to this later

Databricks Free Edition uses serverless compute, so there are no running processes to stop. Your data stays in the workspace at no cost.

  • Close your browser tab when you are done. Nothing continues running in the background.
  • When you return, sign back in to your workspace and reopen your notebook. All tables and data will be exactly as you left them.

Delete - I don't want to use this again

Remove all project data by dropping the catalog. This deletes every schema, table, and volume inside it in one command.

  • Open a new cell in your notebook and run the following command to drop the entire catalog:
spark.sql("DROP CATALOG medallion_project CASCADE")

What does CASCADE do?

The CASCADE keyword tells Databricks to delete the catalog and everything inside it. This removes the bronze, silver, and gold schemas, all four Delta tables, and the managed volume in one go.

  • Verify the catalog is gone by running:
spark.sql("SHOW CATALOGS").show()

You should no longer see medallion_project in the output.

  • To delete your notebook, right-click it in the workspace sidebar and select the delete option.

Your Databricks Free Edition workspace remains active even after deleting all project data. The workspace itself incurs no charges.

Nice Work!

Nice Work!

You just built a production-style medallion data pipeline in Databricks that transforms raw e-commerce order data through bronze, silver, and gold layers. You even broke your own pipeline on purpose and recovered it with Delta Lake time travel.

You've learned how to:

  • Design a three-layer medallion architecture (bronze, silver, gold) governed by Unity Catalog, ingesting raw data into bronze, cleaning and quarantining bad records in silver, and aggregating business metrics in gold.
  • Deliberately corrupt a data pipeline and recover it using Delta Lake time travel, rolling back the silver table to a clean version and rebuilding gold without data loss.
  • Build an incremental MERGE-based upsert pipeline using DeltaTable.merge() to handle new and updated orders without creating duplicates, and enforce schema constraints like NOT NULL to protect data quality at the table level.
  • Secret Mission: Added CHECK constraints and built a data freshness monitor for production-readiness.

Ready to quiz yourself?