Medallion Lakehouse with Delta Lake

Build a Bronze/Silver/Gold data pipeline using PySpark and Delta Lake locally.

Introduction

30 Second Summary

Every company collects data from dozens of sources, but raw data is messy and unreliable. The difference between a junior data engineer and a senior one is knowing how to organize that chaos into layers that business teams can actually trust.

In this project, you will build a complete medallion lakehouse pipeline on your local machine using PySpark and Delta Lake. You'll ingest raw e-commerce order data, clean and validate it with change tracking, and produce business-ready aggregation tables for analytics.

What You'll Build

A local medallion architecture data pipeline that processes e-commerce orders through Bronze, Silver, and Gold layers using Delta Lake for transactional storage.

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

  • A Bronze Delta table that ingests raw CSV data with metadata columns tracking ingestion time, source file, and batch ID.
  • A Silver Delta table with cleaned, deduplicated data and MERGE-based upserts for SCD Type 2 change tracking.
  • A Gold Delta table containing aggregated business metrics like daily revenue by category and product performance.
  • A pipeline orchestrator that runs the full Bronze to Silver to Gold flow and demonstrates Delta Lake time travel and table maintenance.
  • Secret Mission: Add data quality constraints with CHECK enforcement and implement incremental ingestion to simulate a production batch pipeline.

Are there any prerequisites?

This project assumes you have basic PySpark knowledge (reading and writing DataFrames, SQL operations) and Python fundamentals.

You also need VS Code, Python 3.10, and Apache Spark 4.1.1 already installed on your machine. Step 1 covers installing the Delta Lake package.

Set Up Delta Lake and Project Structure

Before you can build a pipeline, you need the storage engine that makes it reliable.

Delta Lake adds ACID transactions, schema enforcement, and time travel on top of Apache Spark. In this step, you will install Delta Lake, scaffold your project folders, and generate the e-commerce data your pipeline will process.

In this step, get ready to:
  • Install delta-spark and verify it works with your Spark installation.
  • Create the project folder structure and generate seed data.
  • Create a reusable SparkSession configuration module.
Install delta-spark and verify compatibility

The delta-spark package is the Python API for Delta Lake. It automatically downloads the required Delta Lake JARs and integrates with your existing Spark installation.

  • Install delta-spark by running this command:
pip install delta-spark==4.3.0

Why this specific version?

Version 4.3.0 requires pyspark<=4.1.1,>=4.0.1, which matches your Spark 4.1.1 installation exactly.

  • You should see pip download and install the package successfully. Look for a line containing Successfully installed delta-spark-4.3.0 in the output.

Seeing a version conflict?

If pip reports a conflict with your PySpark version, confirm you have Spark 4.1.1 by running spark-submit --version. Delta-spark 4.3.0 is not compatible with PySpark 4.1.2 or higher.

Help me fix a delta-spark version conflict with my PySpark installation.

Now create your project folder and open it in VS Code.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir medallion-lakehouse
cd medallion-lakehouse
  • Open this folder in VS Code by running:
code .
  • Confirm the folder exists by running:
ls
  • You should see an empty directory (no output from ls yet since the folder is brand new).

Next, create a quick test script to verify Delta Lake works with your Spark installation.

  • In VS Code, create a new file called test_delta.py in the root of your medallion-lakehouse folder.
  • Paste the following test code into test_delta.py:
import pyspark
from delta import configure_spark_with_delta_pip

# Create a Delta Lake-enabled SparkSession
builder = (
    pyspark.sql.SparkSession.builder.appName("DeltaTest")
    .master("local[*]")
    .config(
        "spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension"
    )
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    )
)
spark = configure_spark_with_delta_pip(builder).getOrCreate()

# Write a small test DataFrame in Delta format
data = [("Alice", 1), ("Bob", 2), ("Charlie", 3)]
df = spark.createDataFrame(data, ["name", "id"])
df.write.format("delta").mode("overwrite").save("test_delta_output")

# Read it back to confirm Delta Lake works
result = spark.read.format("delta").load("test_delta_output")
result.show()

print("Delta Lake is working correctly!")
spark.stop()

What does this code do?

This script creates a SparkSession configured with Delta Lake extensions, writes a tiny DataFrame in Delta format, and reads it back. If the read succeeds and prints a table, Delta Lake is properly connected to your Spark installation.

The configure_spark_with_delta_pip utility automatically downloads the required Delta Lake JARs from Maven so you do not need to manage them manually.

  • Save test_delta.py.
  • Open the integrated terminal in VS Code by pressing Cmd+` (macOS) or Ctrl+` (Windows).
  • Run the test script by executing:
python test_delta.py
  • You should see a table printed with three rows (Alice, Bob, Charlie) followed by the message Delta Lake is working correctly!.

Seeing Java or class-not-found errors?

Delta Lake 4.x requires Java 17 or higher. Run java -version and confirm you see version 17+.

If you see a ModuleNotFoundError for delta, confirm you installed delta-spark in the same Python environment that your terminal is using.

Help me debug my Delta Lake test script errors.

  • Clean up the test artifacts by running:
rm -rf test_delta_output test_delta.py
Create project structure and seed data

A medallion lakehouse organizes data into layers: Bronze (raw), Silver (cleaned), and Gold (aggregated). Your folder structure mirrors these layers so each Delta table lives in a clear, predictable location.

  • Create all the project directories by running this command from your medallion-lakehouse folder:
mkdir -p data/raw data/bronze data/silver data/gold src notebooks
  • Confirm the directories exist by running:
ls
  • You should see data, notebooks, and src listed in the output.

What is each directory for?

  • data/raw/ stores incoming CSV files (your raw source data).
  • data/bronze/ stores the Bronze Delta table (raw data with metadata).
  • data/silver/ stores the Silver Delta table (cleaned and conformed data).
  • data/gold/ stores Gold Delta tables (business-ready aggregations).
  • src/ holds all Python pipeline scripts.
  • notebooks/ is reserved for exploratory Jupyter notebooks.

Now you need realistic data to feed through your pipeline. You will create a script that generates e-commerce order data in two batches to simulate an initial load and a subsequent incremental update.

  • In VS Code, create a new file called src/generate_data.py.
  • Paste the following imports and product data into generate_data.py:
import csv
import os
import random
from datetime import datetime, timedelta

PRODUCTS = [
    ("Wireless Mouse", "Electronics", 29.99),
    ("USB-C Cable", "Electronics", 12.99),
    ("Notebook A5", "Stationery", 5.49),
    ("Ballpoint Pen Pack", "Stationery", 3.99),
    ("Desk Lamp", "Home Office", 45.00),
    ("Monitor Stand", "Home Office", 89.99),
    ("Webcam HD", "Electronics", 59.99),
    ("Sticky Notes", "Stationery", 2.49),
    ("Ergonomic Chair", "Furniture", 299.99),
    ("Standing Desk Mat", "Furniture", 49.99),
]

STATUSES = ["completed", "pending", "shipped", "cancelled"]

What does this set up?

The PRODUCTS list defines 10 realistic products across 4 categories, each with a fixed unit price. The STATUSES list gives each order a random lifecycle state. These become the raw data your Bronze layer will ingest.

  • Add the generate_orders function below the STATUSES list:

def generate_orders(num_orders, start_id, date_range_days=30):
    orders = []
    base_date = datetime(2026, 6, 1)
    for i in range(num_orders):
        product_name, category, unit_price = random.choice(PRODUCTS)
        order_date = base_date + timedelta(days=random.randint(0, date_range_days))
        orders.append(
            {
                "order_id": start_id + i,
                "customer_id": random.randint(1000, 1099),
                "product_name": product_name,
                "category": category,
                "quantity": random.randint(1, 10),
                "unit_price": unit_price,
                "order_date": order_date.strftime("%Y-%m-%d"),
                "status": random.choice(STATUSES),
            }
        )
    return orders

What does this function do?

Each call to generate_orders creates a batch of orders with sequential IDs, random products, random quantities, and dates spread over 30 days. The start_id parameter lets you generate separate batches without overlapping order IDs.

  • Add the write_csv helper and main function below generate_orders:

def write_csv(orders, filepath):
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    with open(filepath, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=orders[0].keys())
        writer.writeheader()
        writer.writerows(orders)
    print(f"Written {len(orders)} orders to {filepath}")


def main():
    random.seed(42)

    batch_1 = generate_orders(num_orders=200, start_id=1)
    write_csv(batch_1, "data/raw/orders_batch_1.csv")

    batch_2_new = generate_orders(num_orders=50, start_id=201)
    batch_2_updates = []
    for order in random.sample(batch_1, 30):
        updated = order.copy()
        updated["status"] = "completed"
        updated["quantity"] = order["quantity"] + 1
        batch_2_updates.append(updated)

    batch_2 = batch_2_updates + batch_2_new
    random.shuffle(batch_2)
    write_csv(batch_2, "data/raw/orders_batch_2.csv")


if __name__ == "__main__":
    main()

What does main() do?

The main function generates two batches that simulate real-world data arrival. Batch 1 is a full initial load of 200 orders.

Batch 2 simulates an incremental update: 30 existing orders with their status changed to completed and quantity bumped, plus 50 brand-new orders. This is exactly the kind of mixed update your Silver layer's MERGE logic will handle later.

  • Save src/generate_data.py.
  • Run the data generation script from the medallion-lakehouse folder:
python src/generate_data.py
  • You should see two confirmation messages printed:
Written 200 orders to data/raw/orders_batch_1.csv
Written 80 orders to data/raw/orders_batch_2.csv
  • Verify the CSV files were created by running:
ls data/raw/
  • You should see orders_batch_1.csv and orders_batch_2.csv listed.

Seeing a FileNotFoundError?

Make sure you are running the command from the medallion-lakehouse root folder, not from inside src/. The script writes to data/raw/ relative to where you run it.

Help me fix a FileNotFoundError when running generate_data.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/generate_data.py and both CSV files exist in data/raw/.

ⓧ I'd like to double check the full code

import csv
import os
import random
from datetime import datetime, timedelta

PRODUCTS = [
    ("Wireless Mouse", "Electronics", 29.99),
    ("USB-C Cable", "Electronics", 12.99),
    ("Notebook A5", "Stationery", 5.49),
    ("Ballpoint Pen Pack", "Stationery", 3.99),
    ("Desk Lamp", "Home Office", 45.00),
    ("Monitor Stand", "Home Office", 89.99),
    ("Webcam HD", "Electronics", 59.99),
    ("Sticky Notes", "Stationery", 2.49),
    ("Ergonomic Chair", "Furniture", 299.99),
    ("Standing Desk Mat", "Furniture", 49.99),
]

STATUSES = ["completed", "pending", "shipped", "cancelled"]


def generate_orders(num_orders, start_id, date_range_days=30):
    orders = []
    base_date = datetime(2026, 6, 1)
    for i in range(num_orders):
        product_name, category, unit_price = random.choice(PRODUCTS)
        order_date = base_date + timedelta(days=random.randint(0, date_range_days))
        orders.append(
            {
                "order_id": start_id + i,
                "customer_id": random.randint(1000, 1099),
                "product_name": product_name,
                "category": category,
                "quantity": random.randint(1, 10),
                "unit_price": unit_price,
                "order_date": order_date.strftime("%Y-%m-%d"),
                "status": random.choice(STATUSES),
            }
        )
    return orders


def write_csv(orders, filepath):
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    with open(filepath, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=orders[0].keys())
        writer.writeheader()
        writer.writerows(orders)
    print(f"Written {len(orders)} orders to {filepath}")


def main():
    random.seed(42)

    batch_1 = generate_orders(num_orders=200, start_id=1)
    write_csv(batch_1, "data/raw/orders_batch_1.csv")

    batch_2_new = generate_orders(num_orders=50, start_id=201)
    batch_2_updates = []
    for order in random.sample(batch_1, 30):
        updated = order.copy()
        updated["status"] = "completed"
        updated["quantity"] = order["quantity"] + 1
        batch_2_updates.append(updated)

    batch_2 = batch_2_updates + batch_2_new
    random.shuffle(batch_2)
    write_csv(batch_2, "data/raw/orders_batch_2.csv")


if __name__ == "__main__":
    main()
Create the shared SparkSession configuration module

Every script in your pipeline needs a Delta Lake-enabled SparkSession. Instead of repeating the configuration in every file, you will create one shared module that all pipeline scripts import.

  • Create a new file called src/spark_session.py.
  • Paste the following code into spark_session.py:
import pyspark
from delta import configure_spark_with_delta_pip


def get_spark_session(app_name="MedallionLakehouse"):
    builder = (
        pyspark.sql.SparkSession.builder.appName(app_name)
        .master("local[*]")
        .config(
            "spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension"
        )
        .config(
            "spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog",
        )
    )
    spark = configure_spark_with_delta_pip(builder).getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    return spark

What does this module do?

  • local[*] tells Spark to use all available CPU cores on your machine.
  • The two .config() lines register Delta Lake's SQL extensions and catalog so Spark knows how to read and write Delta format.
  • configure_spark_with_delta_pip wraps the builder and handles JAR downloads automatically.
  • setLogLevel("WARN") suppresses verbose Spark INFO logs so your pipeline output stays readable.
  • Save src/spark_session.py.
  • Verify the module imports correctly by running this quick test:
cd src && python -c "from spark_session import get_spark_session; print('Import successful')" && cd ..
  • You should see Import successful printed in the terminal.

Seeing a ModuleNotFoundError for delta?

Confirm that delta-spark is installed in the same Python environment by running pip show delta-spark. If nothing shows, re-run the install command from earlier in this step.

Help me fix the delta import error in spark_session.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/spark_session.py and the import test passed.

ⓧ I'd like to double check the full code

import pyspark
from delta import configure_spark_with_delta_pip


def get_spark_session(app_name="MedallionLakehouse"):
    builder = (
        pyspark.sql.SparkSession.builder.appName(app_name)
        .master("local[*]")
        .config(
            "spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension"
        )
        .config(
            "spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog",
        )
    )
    spark = configure_spark_with_delta_pip(builder).getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    return spark

Your environment is fully set up. You have Delta Lake installed, a clean project structure, realistic seed data, and a reusable SparkSession module. Next up, you will build the Bronze layer to ingest that raw CSV data into a proper Delta table with full transaction logging.

Build the Bronze Layer

Your project structure and seed data are ready. Now it's time to build the first layer of the medallion architecture: the Bronze layer.

The Bronze layer is the raw ingestion point where data enters your lakehouse exactly as it arrives from the source. The key difference between Bronze and just dumping CSV files into a folder is that Bronze stores data in Delta Lake format, giving you ACID transactions, schema tracking, and a full audit trail of every write operation.

In this step, get ready to:
  • Ingest raw CSV data into a Bronze Delta table with metadata columns.
  • Explore the Delta Lake transaction log to confirm ACID logging is active.
Ingest raw CSV data into a Bronze Delta table

Every record that enters the Bronze layer needs context about when and how it arrived. You will add metadata columns to track ingestion timestamps, source file names, and batch identifiers.

  • Create a new file called src/bronze_ingestion.py in VS Code.
  • Add the imports and the ingest_to_bronze function by pasting this code:
from pyspark.sql import functions as F
from spark_session import get_spark_session


def ingest_to_bronze(spark, raw_path, bronze_path, batch_id):
    print(f"--- Bronze Ingestion: batch_id={batch_id} ---")

    # Read the raw CSV file with headers and auto-detected types
    df = (
        spark.read.option("header", "true")
        .option("inferSchema", "true")
        .csv(raw_path)
    )

    # Add metadata columns to track data lineage
    df_with_metadata = df.withColumns(
        {
            "_ingestion_timestamp": F.current_timestamp(),
            "_source_file": F.input_file_name(),
            "_batch_id": F.lit(batch_id),
        }
    )

    # Write to the Bronze Delta table in append mode
    df_with_metadata.write.format("delta").mode("append").save(bronze_path)

    # Read back and print the total row count
    count = spark.read.format("delta").load(bronze_path).count()
    print(f"Bronze table now has {count} total rows")
    return count

What does this code do?

This function reads a CSV file into a Spark DataFrame, then enriches it with three metadata columns before writing to Delta format.

  • _ingestion_timestamp records exactly when each row was ingested. This lets you audit when data entered the pipeline.
  • _source_file captures which file the row came from, using PySpark's built-in input_file_name() function.
  • _batch_id tags the entire load with a human-readable identifier so you can trace runs.
  • withColumns accepts a dictionary of column names to expressions, adding all three columns in a single call.
  • Now add the if __name__ block below the function to make the script runnable:
if __name__ == "__main__":
    spark = get_spark_session()
    ingest_to_bronze(
        spark,
        raw_path="data/raw/orders_batch_1.csv",
        bronze_path="data/bronze/orders",
        batch_id="batch_001",
    )
    spark.stop()

What does this block do?

This is the entry point when you run the script directly. It initializes a Delta-enabled SparkSession using your shared get_spark_session() helper, ingests orders_batch_1.csv into the Bronze table at data/bronze/orders, and then cleanly shuts down Spark.

  • Save src/bronze_ingestion.py.
  • Run the Bronze ingestion script from your project folder by running this command:
python src/bronze_ingestion.py

You should see output showing the Bronze ingestion starting, followed by a message confirming the table now has 200 total rows.

Seeing a ModuleNotFoundError?

  • If you see ModuleNotFoundError: No module named 'spark_session', make sure you are running the command from your project's root folder (the one containing the src/ directory). The import relies on running from within src/ or having the correct working directory. Try running cd src && python bronze_ingestion.py instead.
  • If you see a Java-related error, confirm your Java 17 installation is active by running java -version in your terminal.

Help me debug my Bronze ingestion script

✔️ Awesome, I've got everything!

Great. Double check you have saved src/bronze_ingestion.py before moving on.

ⓧ I'd like to double check the full code

from pyspark.sql import functions as F
from spark_session import get_spark_session


def ingest_to_bronze(spark, raw_path, bronze_path, batch_id):
    print(f"--- Bronze Ingestion: batch_id={batch_id} ---")

    df = (
        spark.read.option("header", "true")
        .option("inferSchema", "true")
        .csv(raw_path)
    )

    df_with_metadata = df.withColumns(
        {
            "_ingestion_timestamp": F.current_timestamp(),
            "_source_file": F.input_file_name(),
            "_batch_id": F.lit(batch_id),
        }
    )

    df_with_metadata.write.format("delta").mode("append").save(bronze_path)

    count = spark.read.format("delta").load(bronze_path).count()
    print(f"Bronze table now has {count} total rows")
    return count


if __name__ == "__main__":
    spark = get_spark_session()
    ingest_to_bronze(
        spark,
        raw_path="data/raw/orders_batch_1.csv",
        bronze_path="data/bronze/orders",
        batch_id="batch_001",
    )
    spark.stop()
Explore the Delta Lake transaction log

Every write to a Delta table is recorded as a versioned commit in the transaction log. This log lives in a _delta_log/ folder inside your table directory and is what gives Delta Lake its ACID guarantees.

  • Verify the transaction log exists by running:
ls data/bronze/orders/_delta_log/

You should see at least one JSON file (like 00000000000000000000.json) representing the first commit.

  • Now explore the table programmatically. Run this command to verify the schema, row count, and transaction history:
cd src && python -c "
from delta.tables import DeltaTable
from spark_session import get_spark_session

spark = get_spark_session()
bronze_path = 'data/bronze/orders'

# Read back the Bronze table and show its schema
bronze_df = spark.read.format('delta').load(bronze_path)
print('Schema:')
bronze_df.printSchema()
print(f'Row count: {bronze_df.count()}')

# Access the transaction log history
bronze_table = DeltaTable.forPath(spark, bronze_path)
print('\nTransaction History:')
bronze_table.history().select('version', 'timestamp', 'operation').show(truncate=False)

spark.stop()
"

What should I see?

The schema printout shows all 8 original columns (order_id, customer_id, product_name, category, quantity, unit_price, order_date, status) plus your 3 metadata columns (_ingestion_timestamp, _source_file, _batch_id). The row count is 200. The transaction history table shows version 0 with operation WRITE and a timestamp.

This single history entry proves that Delta Lake is recording every write as an ACID transaction. In later steps, you will see this log grow with each pipeline run.

Seeing 'not a Delta table' error?

  • Make sure the previous bronze_ingestion.py ran successfully and printed 200 rows. If it failed silently, the Delta table was never created.
  • Check that the path data/bronze/orders exists and contains both Parquet data files and a _delta_log folder.

Help me fix my Delta Lake transaction log exploration

What is a transaction log?

The _delta_log folder stores a JSON file for every commit (write operation) to your table. Each file records which data files were added or removed, the schema at that point, and operation metadata.

This is what separates Delta Lake from plain Parquet. Without the log, concurrent reads and writes can produce corrupted results. With it, you get database-level reliability on top of file storage.

Your Bronze layer is operational. Raw data flows in with full lineage tracking, and every write is recorded as a versioned ACID transaction. Next up, you will build the Silver layer to clean and deduplicate this data using Delta Lake's powerful MERGE operation.

Build the Silver Layer

Your Bronze layer stores raw data exactly as it arrived. That's useful for auditability, but raw data contains problems: invalid values, duplicate records, and inconsistent formatting.

The Silver layer is where you clean, validate, and conform data into a trusted, query-ready state. In this step, you'll build a Silver transformation that enforces data quality rules and uses Delta Lake's MERGE operation to track how records change over time.

In this step, get ready to:
  • Apply data quality rules to filter, deduplicate, and standardize Bronze data.
  • Implement a MERGE (upsert) operation for SCD Type 2 change tracking.
  • Verify that updated records are tracked separately from newly inserted ones.
Apply data quality transformations

Raw data in the Bronze table may contain orders with zero or negative quantities, duplicate rows from overlapping ingestion batches, and inconsistent casing in the status column. The Silver layer needs to reject bad records and standardize the rest before any downstream analytics can trust it.

  • Create a new file called silver_transformation.py inside your src/ folder.
  • Add the imports, function definition, and data quality transformation logic by pasting this code:
from delta.tables import DeltaTable
from pyspark.sql import functions as F
from spark_session import get_spark_session


def transform_to_silver(spark, bronze_path, silver_path):
    print("--- Silver Transformation ---")

    # Read all rows from the Bronze Delta table
    bronze_df = spark.read.format("delta").load(bronze_path)
    print(f"Read {bronze_df.count()} rows from Bronze")

    # Apply data quality rules and compute total_amount
    cleaned_df = (
        bronze_df.filter(F.col("quantity") > 0)
        .filter(F.col("unit_price") > 0)
        .dropDuplicates(["order_id", "_batch_id"])
        .withColumn("status", F.upper(F.col("status")))
        .withColumn("total_amount", F.col("quantity") * F.col("unit_price"))
    )

    print(f"After cleaning: {cleaned_df.count()} rows")

    # Select only the columns needed for the Silver table
    silver_df = cleaned_df.select(
        "order_id",
        "customer_id",
        "product_name",
        "category",
        "quantity",
        "unit_price",
        "total_amount",
        "order_date",
        "status",
    )

What does this code do?

This code reads from the Bronze Delta table and applies four data quality rules in sequence:

  • filter(F.col("quantity") > 0) and filter(F.col("unit_price") > 0) remove rows with invalid values that would corrupt downstream calculations.
  • dropDuplicates(["order_id", "_batch_id"]) eliminates exact duplicate rows that can appear when ingestion batches overlap.
  • F.upper(F.col("status")) standardizes the status column to uppercase so comparisons downstream are consistent.
  • The total_amount computed column multiplies quantity by unit price, saving the Gold layer from recalculating it every time.

Why clean data at the Silver layer instead of Bronze?

Bronze stores data exactly as it arrived, preserving the raw source for auditing and reprocessing. If you cleaned at ingestion time, you'd lose the original records and couldn't rerun transformations with different rules later.

Silver is where you enforce your opinion about data quality. This separation is the core design principle of the medallion architecture.

  • Save src/silver_transformation.py. You won't run it yet because the function isn't complete.
Implement MERGE (upsert) for SCD Type 2 tracking

In production pipelines, new data batches often contain a mix of brand-new records and updates to existing ones. A naive overwrite would lose the history of when records were first created versus when they changed.

Delta Lake's MERGE operation solves this by matching incoming rows against existing ones and choosing to insert or update based on conditions. This pattern is called SCD Type 2 (Slowly Changing Dimension), and it tracks exactly when each record was created and last modified.

  • In src/silver_transformation.py, add the MERGE logic and the main block directly below the silver_df = cleaned_df.select(...) statement you just wrote:
    # First run: no Silver table exists, so write directly
    # Subsequent runs: MERGE new/changed records into existing Silver table
    if DeltaTable.isDeltaTable(spark, silver_path):
        silver_table = DeltaTable.forPath(spark, silver_path)
        (
            silver_table.alias("target")
            .merge(silver_df.alias("source"), "target.order_id = source.order_id")
            .whenMatchedUpdate(
                condition="target.status != source.status OR target.quantity != source.quantity",
                set={
                    "customer_id": "source.customer_id",
                    "product_name": "source.product_name",
                    "category": "source.category",
                    "quantity": "source.quantity",
                    "unit_price": "source.unit_price",
                    "total_amount": "source.total_amount",
                    "order_date": "source.order_date",
                    "status": "source.status",
                    "_updated_at": F.current_timestamp(),
                },
            )
            .whenNotMatchedInsert(
                values={
                    "order_id": "source.order_id",
                    "customer_id": "source.customer_id",
                    "product_name": "source.product_name",
                    "category": "source.category",
                    "quantity": "source.quantity",
                    "unit_price": "source.unit_price",
                    "total_amount": "source.total_amount",
                    "order_date": "source.order_date",
                    "status": "source.status",
                    "_created_at": F.current_timestamp(),
                    "_updated_at": F.lit(None).cast("timestamp"),
                }
            )
            .execute()
        )
    else:
        # First run: create the Silver table with audit columns
        silver_df_with_audit = silver_df.withColumns(
            {
                "_created_at": F.current_timestamp(),
                "_updated_at": F.lit(None).cast("timestamp"),
            }
        )
        silver_df_with_audit.write.format("delta").mode("overwrite").save(silver_path)

    final_count = spark.read.format("delta").load(silver_path).count()
    print(f"Silver table now has {final_count} rows")
    return final_count


if __name__ == "__main__":
    spark = get_spark_session()
    transform_to_silver(spark, "data/bronze/orders", "data/silver/orders")
    spark.stop()

What does the MERGE logic do?

DeltaTable.isDeltaTable(spark, silver_path) checks whether the Silver table already exists. On the very first run, it doesn't, so the else branch writes the cleaned data directly with _created_at and _updated_at audit columns.

On subsequent runs, the MERGE matches incoming rows to existing ones by order_id. If a matching record's status or quantity has changed, it updates the row and stamps _updated_at. If no match exists, it inserts the record with a fresh _created_at.

  • Save src/silver_transformation.py.
  • Run the Silver transformation for the first time by running this command from your project folder:
python src/silver_transformation.py

You should see output showing 200 rows read from Bronze, a count after cleaning, and the Silver table created with 200 rows.

Seeing a ModuleNotFoundError?

Make sure you're running the command from your project's root folder (the one containing the src/ directory). PySpark needs to find spark_session.py in the same directory as the script you're running.

If you get an error about spark_session not being found, try running from inside the src/ directory instead:

cd src
python silver_transformation.py

Help me debug my Silver transformation script.

Now you'll simulate a real-world scenario where a second batch of data arrives. This batch contains updates to 30 existing orders and 50 brand-new orders.

  • Ingest the second batch into Bronze by running:
python -c "import sys; sys.path.insert(0, 'src'); from spark_session import get_spark_session; from bronze_ingestion import ingest_to_bronze; spark = get_spark_session(); ingest_to_bronze(spark, 'data/raw/orders_batch_2.csv', 'data/bronze/orders', 'batch_002'); spark.stop()"

What does this command do?

This one-liner imports your existing ingest_to_bronze function and runs it with orders_batch_2.csv as the source. After it runs, your Bronze table will contain 280 total rows (200 from batch 1 plus 80 from batch 2).

  • Run the Silver transformation again to trigger the MERGE path:
python src/silver_transformation.py

This time, the script detects the existing Silver table and runs a MERGE instead of an overwrite. You should see 280 rows read from Bronze and the Silver table ending with 250 rows.

Why 250 rows and not 280?

Bronze stores every record from every batch (280 total). But the Silver layer deduplicates by order_id via MERGE. The 30 updated orders in batch 2 matched existing rows and were updated in place rather than duplicated. The 50 new orders were inserted. So Silver has 200 + 50 = 250 unique orders.

Silver table still shows 200 rows?

Make sure you ran the Bronze ingestion for batch 2 before re-running the Silver script. The Silver transformation reads from Bronze, so if batch 2 isn't in Bronze yet, the MERGE has no new data to process.

Help me debug why my Silver MERGE isn't picking up new records.

✔️ Awesome, I've got everything!

Great. Double check you've saved src/silver_transformation.py and that your Silver table has 250 rows after the second run.

ⓧ I'd like to double check the full code

Here's the complete src/silver_transformation.py file:

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


def transform_to_silver(spark, bronze_path, silver_path):
    print("--- Silver Transformation ---")

    bronze_df = spark.read.format("delta").load(bronze_path)
    print(f"Read {bronze_df.count()} rows from Bronze")

    cleaned_df = (
        bronze_df.filter(F.col("quantity") > 0)
        .filter(F.col("unit_price") > 0)
        .dropDuplicates(["order_id", "_batch_id"])
        .withColumn("status", F.upper(F.col("status")))
        .withColumn("total_amount", F.col("quantity") * F.col("unit_price"))
    )

    print(f"After cleaning: {cleaned_df.count()} rows")

    silver_df = cleaned_df.select(
        "order_id",
        "customer_id",
        "product_name",
        "category",
        "quantity",
        "unit_price",
        "total_amount",
        "order_date",
        "status",
    )

    if DeltaTable.isDeltaTable(spark, silver_path):
        silver_table = DeltaTable.forPath(spark, silver_path)
        (
            silver_table.alias("target")
            .merge(silver_df.alias("source"), "target.order_id = source.order_id")
            .whenMatchedUpdate(
                condition="target.status != source.status OR target.quantity != source.quantity",
                set={
                    "customer_id": "source.customer_id",
                    "product_name": "source.product_name",
                    "category": "source.category",
                    "quantity": "source.quantity",
                    "unit_price": "source.unit_price",
                    "total_amount": "source.total_amount",
                    "order_date": "source.order_date",
                    "status": "source.status",
                    "_updated_at": F.current_timestamp(),
                },
            )
            .whenNotMatchedInsert(
                values={
                    "order_id": "source.order_id",
                    "customer_id": "source.customer_id",
                    "product_name": "source.product_name",
                    "category": "source.category",
                    "quantity": "source.quantity",
                    "unit_price": "source.unit_price",
                    "total_amount": "source.total_amount",
                    "order_date": "source.order_date",
                    "status": "source.status",
                    "_created_at": F.current_timestamp(),
                    "_updated_at": F.lit(None).cast("timestamp"),
                }
            )
            .execute()
        )
    else:
        silver_df_with_audit = silver_df.withColumns(
            {
                "_created_at": F.current_timestamp(),
                "_updated_at": F.lit(None).cast("timestamp"),
            }
        )
        silver_df_with_audit.write.format("delta").mode("overwrite").save(silver_path)

    final_count = spark.read.format("delta").load(silver_path).count()
    print(f"Silver table now has {final_count} rows")
    return final_count


if __name__ == "__main__":
    spark = get_spark_session()
    transform_to_silver(spark, "data/bronze/orders", "data/silver/orders")
    spark.stop()

Your Silver layer is now live with clean, deduplicated, and change-tracked data. Next up, you'll build the Gold layer to turn this trusted dataset into business-ready aggregated tables that answer real analytics questions.

Build the Gold Layer

Your Silver table now holds 250 clean, validated order records with full change tracking.

The final layer of the medallion architecture is Gold. Gold tables are purpose-built for business analysts and dashboards. Instead of storing individual records, they hold pre-computed aggregations like daily revenue and product performance metrics. You will also use Delta Lake time travel to query historical versions of your Silver table, proving that every past state is recoverable.

In this step, get ready to:
  • Create aggregated Gold tables for daily revenue and product performance.
  • Demonstrate Delta Lake time travel by querying historical versions of the Silver table.
Create aggregated business tables

Gold tables answer specific business questions. Rather than forcing analysts to scan 250 individual orders, you pre-compute the numbers they care about: revenue by day, and performance by product.

  • Create a new file called gold_aggregation.py inside your src/ folder.
  • Add the following code to build both Gold tables:
from pyspark.sql import functions as F
from spark_session import get_spark_session


def build_gold_tables(spark, silver_path, gold_base_path):
    print("--- Gold Aggregation ---")

    # Read the cleaned Silver table as our source of truth
    silver_df = spark.read.format("delta").load(silver_path)

    # Gold table 1: Daily revenue grouped by date and category
    daily_revenue = silver_df.groupBy("order_date", "category").agg(
        F.sum("total_amount").alias("total_revenue"),
        F.count("order_id").alias("order_count"),
    )

    daily_revenue_path = f"{gold_base_path}/daily_revenue"
    daily_revenue.write.format("delta").mode("overwrite").save(daily_revenue_path)
    print(f"Gold daily_revenue: {daily_revenue.count()} rows")

    # Gold table 2: Product-level performance metrics
    product_performance = silver_df.groupBy("product_name", "category").agg(
        F.sum("total_amount").alias("total_revenue"),
        F.sum("quantity").alias("total_quantity_sold"),
        F.avg("total_amount").alias("avg_order_value"),
        F.count("order_id").alias("total_orders"),
    )

    product_performance_path = f"{gold_base_path}/product_performance"
    product_performance.write.format("delta").mode("overwrite").save(
        product_performance_path
    )
    print(f"Gold product_performance: {product_performance.count()} rows")


if __name__ == "__main__":
    spark = get_spark_session()
    build_gold_tables(spark, "data/silver/orders", "data/gold")
    spark.stop()

What does this code do?

The build_gold_tables function reads the Silver table and produces two separate Gold outputs.

  • daily_revenue groups orders by order_date and category, then sums up revenue and counts orders per group.
  • product_performance groups by product_name to calculate total revenue, quantity sold, average order value, and total orders per product.
  • Both use overwrite mode because Gold tables are always recomputed from Silver. If the underlying data changes, Gold reflects those changes on the next run.
  • Save src/gold_aggregation.py.
  • Run the Gold aggregation script by running this command from your src/ directory:
python gold_aggregation.py

You should see output showing the row counts for both Gold tables, similar to:

--- Gold Aggregation ---
Gold daily_revenue: <number> rows
Gold product_performance: 10 rows

Why 10 rows for product performance?

The seed data has exactly 10 distinct products. Each product gets one row in the performance table, regardless of how many orders it has.

Seeing a ModuleNotFoundError?

Make sure you are running the script from inside the src/ directory so Python can find spark_session.py. If you see an error about the Silver table path not existing, confirm that you ran silver_transformation.py in the previous step.

help me debug my Gold aggregation script

Demonstrate Delta Lake time travel

Every write to a Delta table creates a new version in its transaction log. Time travel lets you query any previous version of the table, giving you a complete audit trail of how your data evolved.

Your Silver table has at least two versions: version 0 from the initial batch 1 load, and the latest version after the MERGE with batch 2. Let's query both to see the difference.

  • Run this time travel demonstration script from your src/ directory:
# Run from src/ directory:
# python -c "exec(open('time_travel_demo.py').read())"
# Or save as src/time_travel_demo.py and run: python time_travel_demo.py

from pyspark.sql import functions as F
from spark_session import get_spark_session

spark = get_spark_session()

silver_path = "data/silver/orders"

# Query version 0: the state after the first batch was loaded
silver_v0 = spark.read.format("delta").option("versionAsOf", 0).load(silver_path)
print(f"Silver table version 0: {silver_v0.count()} rows")

# Query the latest version: after MERGE with batch 2
silver_latest = spark.read.format("delta").load(silver_path)
print(f"Silver table latest:     {silver_latest.count()} rows")

# Show how many records were updated via MERGE
updated_records = silver_latest.filter(F.col("_updated_at").isNotNull())
print(f"Records updated via MERGE: {updated_records.count()}")

spark.stop()

What does this code do?

  • spark.read.format("delta").option("versionAsOf", 0).load(path) reads the Silver table exactly as it existed after the very first write. This was the 200-row initial load from batch 1.
  • Reading without versionAsOf gives you the latest version, which contains 250 rows (the original 200 plus 50 new records from batch 2).
  • Filtering for non-null _updated_at reveals the 30 records that were modified during the MERGE operation.
  • Save the file as src/time_travel_demo.py and run it:
python time_travel_demo.py

You should see output showing the difference between versions:

Silver table version 0: 200 rows
Silver table latest:     250 rows
Records updated via MERGE: 30

Why is time travel useful?

In production pipelines, time travel enables data auditing, rollback after bad ingestion, and reproducing past reports. You can always prove what the data looked like at any point in the pipeline's history.

Seeing an AnalysisException about version not found?

This means the Silver table does not have a version 0 in its transaction log. Make sure you ran silver_transformation.py twice in the previous step (once with batch 1, once after batch 2 was ingested into Bronze). Each run creates a new version.

help me troubleshoot Delta Lake time travel queries

✔️ Awesome, I've got everything!

Great. Double check you have saved src/gold_aggregation.py and confirmed both Gold tables output correct row counts.

ⓧ I'd like to double check the full code

Here is the complete src/gold_aggregation.py file:

from pyspark.sql import functions as F
from spark_session import get_spark_session


def build_gold_tables(spark, silver_path, gold_base_path):
    print("--- Gold Aggregation ---")

    silver_df = spark.read.format("delta").load(silver_path)

    daily_revenue = silver_df.groupBy("order_date", "category").agg(
        F.sum("total_amount").alias("total_revenue"),
        F.count("order_id").alias("order_count"),
    )

    daily_revenue_path = f"{gold_base_path}/daily_revenue"
    daily_revenue.write.format("delta").mode("overwrite").save(daily_revenue_path)
    print(f"Gold daily_revenue: {daily_revenue.count()} rows")

    product_performance = silver_df.groupBy("product_name", "category").agg(
        F.sum("total_amount").alias("total_revenue"),
        F.sum("quantity").alias("total_quantity_sold"),
        F.avg("total_amount").alias("avg_order_value"),
        F.count("order_id").alias("total_orders"),
    )

    product_performance_path = f"{gold_base_path}/product_performance"
    product_performance.write.format("delta").mode("overwrite").save(
        product_performance_path
    )
    print(f"Gold product_performance: {product_performance.count()} rows")


if __name__ == "__main__":
    spark = get_spark_session()
    build_gold_tables(spark, "data/silver/orders", "data/gold")
    spark.stop()

Your medallion lakehouse now has all three layers operational. Next up, you will wire Bronze, Silver, and Gold together into a single pipeline orchestrator with table maintenance.

Orchestrate and Optimize the Pipeline

You have three separate scripts that each handle one layer of your medallion architecture. In production, these layers run in sequence as a single coordinated job.

In this step, you will wire Bronze, Silver, and Gold into one orchestrator script that processes both batches, demonstrates Delta Lake time travel, and applies table maintenance to keep your lakehouse healthy.

In this step, get ready to:
  • Create a pipeline orchestrator that runs all three layers end-to-end with logging.
  • Apply OPTIMIZE and VACUUM maintenance operations to compact files and clean up old data.
Create the pipeline orchestrator

Instead of running each layer script by hand, a single orchestrator calls them in the correct order and logs progress at every stage. This is how production data pipelines work.

  • Create a new file called run_pipeline.py inside your src/ folder.
  • Add the imports and the first half of the orchestrator function by pasting this code:
import shutil
from delta.tables import DeltaTable
from pyspark.sql import functions as F
from spark_session import get_spark_session
from bronze_ingestion import ingest_to_bronze
from silver_transformation import transform_to_silver
from gold_aggregation import build_gold_tables


def run_full_pipeline():
    spark = get_spark_session()

    print("=" * 60)
    print("MEDALLION LAKEHOUSE PIPELINE")
    print("=" * 60)

    # Batch 1: initial load through all three layers
    print("\n[1/3] BRONZE LAYER - Raw Ingestion")
    ingest_to_bronze(spark, "data/raw/orders_batch_1.csv", "data/bronze/orders", "batch_001")

    print("\n[2/3] SILVER LAYER - Clean and Conform")
    transform_to_silver(spark, "data/bronze/orders", "data/silver/orders")

    print("\n[3/3] GOLD LAYER - Business Aggregations")
    build_gold_tables(spark, "data/silver/orders", "data/gold")

What does this code do?

The imports pull in each layer function you built in previous steps. The orchestrator calls them in order: Bronze first (raw ingestion), Silver second (cleaning and MERGE), Gold last (aggregations).

Each layer gets a numbered log line so you can track progress. One shared SparkSession is passed to all functions, keeping resource usage efficient.

  • Now add the second half of the function directly below the code you just pasted. This processes batch 2, demonstrates time travel, and closes the script:
    # Batch 2: incremental load simulating new data arriving
    print("\n" + "=" * 60)
    print("PIPELINE COMPLETE - Processing batch 2 (incremental)")
    print("=" * 60)

    print("\n[1/3] BRONZE LAYER - Batch 2 Ingestion")
    ingest_to_bronze(spark, "data/raw/orders_batch_2.csv", "data/bronze/orders", "batch_002")

    print("\n[2/3] SILVER LAYER - Merge new/updated records")
    transform_to_silver(spark, "data/bronze/orders", "data/silver/orders")

    print("\n[3/3] GOLD LAYER - Recompute aggregations")
    build_gold_tables(spark, "data/silver/orders", "data/gold")

    # Demonstrate Delta Lake time travel
    print("\n" + "=" * 60)
    print("TIME TRAVEL DEMONSTRATION")
    print("=" * 60)

    silver_path = "data/silver/orders"
    silver_v0 = spark.read.format("delta").option("versionAsOf", 0).load(silver_path)
    silver_latest = spark.read.format("delta").load(silver_path)
    print(f"Silver table version 0: {silver_v0.count()} rows")
    print(f"Silver table latest:     {silver_latest.count()} rows")

    updated_records = silver_latest.filter(F.col("_updated_at").isNotNull())
    print(f"Records updated via MERGE: {updated_records.count()}")

    print("\nPIPELINE FINISHED SUCCESSFULLY")
    spark.stop()


if __name__ == "__main__":
    run_full_pipeline()

What does the time travel section do?

After both batches are processed, the script reads the Silver table at version 0 (the initial write) and compares it to the latest version. This proves that Delta Lake stores full history of every change.

The versionAsOf option lets you query any past state of a table. Version 0 has 200 rows (batch 1 only). The latest version has 250 rows (after MERGE added 50 new records).

  • Before running the pipeline, clear the existing Delta tables from your previous step-by-step runs. In your terminal, run this from your project root:
rm -rf data/bronze data/silver data/gold
  • Run the pipeline orchestrator by executing:
python src/run_pipeline.py

You should see structured output with section headers for each layer, row counts at every stage, and the time travel comparison at the end showing version 0 with 200 rows and the latest version with 250 rows.

Seeing an import error?

Make sure you are running the command from inside the src/ directory, or that your terminal is in the project root and you use python src/run_pipeline.py. Python needs to find spark_session.py in the same directory.

If you see a "No module named spark_session" error, cd src first, then run python run_pipeline.py.

Help me debug my run_pipeline.py import error

Apply Delta Lake table maintenance

As Delta tables accumulate writes, they produce many small files. OPTIMIZE compacts these into larger, more efficient files. VACUUM removes old files that the transaction log no longer references.

  • In src/run_pipeline.py, find these two lines near the end of the run_full_pipeline() function:
    print("\nPIPELINE FINISHED SUCCESSFULLY")
    spark.stop()
  • Add the following table maintenance section directly above those two lines:
    # Table maintenance: compact small files and remove stale data
    print("\n" + "=" * 60)
    print("TABLE MAINTENANCE")
    print("=" * 60)

    silver_table = DeltaTable.forPath(spark, silver_path)
    silver_table.optimize().executeCompaction()
    print("OPTIMIZE complete")

    spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")
    silver_table.vacuum(0)
    print("VACUUM complete")

    print("\nTransaction History (last 5 entries):")
    silver_table.history(5).select("version", "timestamp", "operation").show(
        truncate=False
    )

What does this maintenance code do?

  • optimize().executeCompaction() merges small Parquet files into larger ones, speeding up future reads.
  • vacuum(0) removes files older than 0 hours that are no longer in the active table version. In production, you would use 168 hours (7 days) to keep time travel working.
  • The retention check is disabled with spark.databricks.delta.retentionDurationCheck.enabled set to false because Delta Lake normally blocks a 0-hour vacuum as a safety measure.
  • history(5) prints the last 5 transaction log entries, confirming OPTIMIZE and VACUUM are recorded.

Why 0-hour retention for this demo?

Production pipelines use 168 hours (7 days) so that concurrent readers and time travel queries still work. Here you use 0 hours purely to demonstrate that VACUUM runs and appears in the transaction log. On a local machine with no concurrent readers, this is safe.

  • Save src/run_pipeline.py.
  • Clear the data directories again so the pipeline runs cleanly from scratch:
rm -rf data/bronze data/silver data/gold
  • Run the full pipeline again to see the maintenance operations in action:
python src/run_pipeline.py

At the end of the output, you should see the transaction history table with entries for WRITE, MERGE, OPTIMIZE, and VACUUM operations. The Silver table still reads correctly after maintenance.

VACUUM throws a retention error?

Make sure the spark.conf.set line appears BEFORE the silver_table.vacuum(0) call. Delta Lake requires the retention check to be disabled before it allows a vacuum with less than the default retention period.

Help me fix my VACUUM retention error in run_pipeline.py

✔️ Awesome, I've got everything!

Great. Make sure you have saved src/run_pipeline.py and that your latest run shows the transaction history table at the end.

ⓧ I'd like to double check the full code

Here is the complete src/run_pipeline.py file:

import shutil
from delta.tables import DeltaTable
from pyspark.sql import functions as F
from spark_session import get_spark_session
from bronze_ingestion import ingest_to_bronze
from silver_transformation import transform_to_silver
from gold_aggregation import build_gold_tables


def run_full_pipeline():
    spark = get_spark_session()

    print("=" * 60)
    print("MEDALLION LAKEHOUSE PIPELINE")
    print("=" * 60)

    print("\n[1/3] BRONZE LAYER - Raw Ingestion")
    ingest_to_bronze(spark, "data/raw/orders_batch_1.csv", "data/bronze/orders", "batch_001")

    print("\n[2/3] SILVER LAYER - Clean and Conform")
    transform_to_silver(spark, "data/bronze/orders", "data/silver/orders")

    print("\n[3/3] GOLD LAYER - Business Aggregations")
    build_gold_tables(spark, "data/silver/orders", "data/gold")

    print("\n" + "=" * 60)
    print("PIPELINE COMPLETE - Processing batch 2 (incremental)")
    print("=" * 60)

    print("\n[1/3] BRONZE LAYER - Batch 2 Ingestion")
    ingest_to_bronze(spark, "data/raw/orders_batch_2.csv", "data/bronze/orders", "batch_002")

    print("\n[2/3] SILVER LAYER - Merge new/updated records")
    transform_to_silver(spark, "data/bronze/orders", "data/silver/orders")

    print("\n[3/3] GOLD LAYER - Recompute aggregations")
    build_gold_tables(spark, "data/silver/orders", "data/gold")

    print("\n" + "=" * 60)
    print("TIME TRAVEL DEMONSTRATION")
    print("=" * 60)

    silver_path = "data/silver/orders"
    silver_v0 = spark.read.format("delta").option("versionAsOf", 0).load(silver_path)
    silver_latest = spark.read.format("delta").load(silver_path)
    print(f"Silver table version 0: {silver_v0.count()} rows")
    print(f"Silver table latest:     {silver_latest.count()} rows")

    updated_records = silver_latest.filter(F.col("_updated_at").isNotNull())
    print(f"Records updated via MERGE: {updated_records.count()}")

    print("\n" + "=" * 60)
    print("TABLE MAINTENANCE")
    print("=" * 60)

    silver_table = DeltaTable.forPath(spark, silver_path)
    silver_table.optimize().executeCompaction()
    print("OPTIMIZE complete")

    spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")
    silver_table.vacuum(0)
    print("VACUUM complete")

    print("\nTransaction History (last 5 entries):")
    silver_table.history(5).select("version", "timestamp", "operation").show(
        truncate=False
    )

    print("\nPIPELINE FINISHED SUCCESSFULLY")
    spark.stop()


if __name__ == "__main__":
    run_full_pipeline()

Your medallion lakehouse pipeline is complete. You have a single script that ingests raw data, cleans and merges it with change tracking, produces business aggregations, proves time travel works, and maintains table health. This is exactly the kind of end-to-end pipeline that data engineering teams run in production on platforms like Databricks and Microsoft Fabric.

Secret mission

Add Data Quality Constraints and Incremental Ingestion

Your pipeline works, but it has two production gaps. Anyone can write bad data into your Silver table, and every run reprocesses files that have already been ingested. In this mission, you'll enforce data quality at the storage layer with CHECK constraints and build an incremental ingestion pattern that tracks which files have already been processed.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on your local machine using open-source tools. There are no cloud resources, subscriptions, or ongoing costs to worry about.

Decide whether to keep your project files on your machine, or delete them entirely.

Resources you used:

  • Project folder on your Desktop containing src/, data/, and notebooks/ directories with all Delta tables and source code.
  • The delta-spark pip package installed in your Python environment.

Keep everything running

No action needed. Choose this if you want to keep experimenting with your medallion pipeline, add new data batches, or extend the project further.

  • Your Delta tables in data/ will stay on disk and remain queryable any time you start a new Spark session.
  • The delta-spark package stays installed and ready to use in other projects.

Pause - I'll come back to this later

Since this project is entirely local with no running processes, there is nothing to pause. Your files sit on disk and use no memory or CPU when idle.

  • Close VS Code to free up memory.
  • Your project folder and all Delta tables remain intact on your Desktop whenever you want to return.

Delete - I don't want to use this again

Remove all project files and uninstall the Delta Lake package to start completely fresh.

  • Close VS Code if it is still open.
  • Delete the project folder by running this command (macOS/Linux):
cd ~/Desktop
rm -rf medallion-lakehouse

On Windows, run this instead:

cd ~\Desktop
Remove-Item -Recurse -Force medallion-lakehouse
  • Uninstall the delta-spark pip package by running:
pip uninstall delta-spark -y
  • Confirm the folder is gone by running ls ~/Desktop (macOS/Linux) or ls ~\Desktop (Windows). You should no longer see medallion-lakehouse in the output.

Nice Work!

Nice Work!

You just built a production-grade medallion lakehouse pipeline using PySpark and Delta Lake. This is the exact architecture running inside enterprise data platforms like Databricks and Microsoft Fabric.

You've learned how to:

  • Build a complete medallion lakehouse pipeline (Bronze, Silver, Gold) with Delta Lake ACID transactions and full transaction log history.
  • Apply MERGE (upsert) operations for SCD Type 2 slowly changing dimension tracking in the Silver layer.
  • Demonstrate Delta Lake time travel to query historical versions of your data and apply table maintenance with OPTIMIZE and VACUUM.
  • Secret Mission: Enforced data quality with CHECK constraints and implemented incremental batch ingestion to simulate a production pipeline.

Ready to quiz yourself?