Databricks Medallion Architecture — Sales Data Pipeline

A hands-on data engineering project built on Databricks and PySpark/Spark SQL, demonstrating the Medallion Architecture (Bronze → Silver → Gold), incremental (delta) loading, a Kimball-style star schema (Dimension & Fact tables), and standalone Slowly Changing Dimension (SCD) Type 1 & Type 2 implementations using MERGE INTO.

This is a learning/portfolio project simulating an e-commerce sales dataset flowing through a lakehouse.

---

Architecture

flowchart LR
    A[(Source Table\ndatamodeling.default.source_data)] -->|Incremental Extract\nWatermark: order_date| B[Bronze Layer\nbronze.bronze_table]
    B -->|Clean & Transform| C[Silver Layer\nsilver.silver_table]
    C -->|Dimensional Modeling| D[Gold Layer\nStar Schema]
    D --> E[Dim Tables]
    D --> F[Fact Table]

Loading

Layer	Purpose	Load Pattern
Source	Raw transactional table simulating an OLTP system	Manual inserts (initial + incremental)
Bronze	Raw ingestion, minimal transformation	Incremental, watermark-based (order_date)
Silver	Cleaned, standardized, deduplicated	Merge (upsert) on order_id
Gold	Business-ready star schema	Full rebuild of dimensions + fact

---

Tech Stack

• Platform: Databricks (Notebooks)
• Processing: PySpark, Spark SQL
• Storage: Delta Lake (Unity Catalog: datamodeling.<layer>.<table>)
• Patterns: Incremental loading, MERGE INTO upserts, ROW_NUMBER() surrogate keys, SCD Type 1 & Type 2

---

Repository Structure

File	Layer	Description
Source.py	Source	Creates source_data table and simulates initial + incremental inserts
Bronze.py	Bronze	Incrementally pulls new rows from source based on max(order_date) watermark
Silver.py	Silver	Cleans/enriches bronze data and merges into the silver table
Gold.py	Gold	Builds dimension tables and the fact table (star schema)
SCDs.py	Standalone Demo	Implements SCD Type 1 (overwrite) and SCD Type 2 (historical tracking) on a separate product dataset

---

Pipeline Details

1. Source Layer (Source.py)

Creates datamodeling.default.source_data, a flat table representing raw order transactions (order, customer, product, payment, and region attributes). Data is inserted in two batches — an initial load (3 orders) and an incremental load (2 more orders) — to simulate new data arriving over time.

2. Bronze Layer (Bronze.py)

Implements incremental extraction using a watermark pattern:

• Checks if bronze.bronze_table already exists.
• If it does, pulls the max(order_date) already loaded; otherwise defaults to 1000-01-01.
• Selects only source rows with order_date greater than that watermark.
• Writes the result with CREATE OR REPLACE TABLE, so each run currently replaces the bronze table rather than appending — worth noting if you extend this into a true incremental history.

3. Silver Layer (Silver.py)

Transforms bronze data by:

• Adding customer_name_upper (standardized customer name).
• Adding processDate (load/processing timestamp).
• Merging into silver.silver_table using MERGE INTO ... ON order_id, with UPDATE SET * on match and INSERT * on no match — a standard upsert pattern.

4. Gold Layer (Gold.py)

Builds a star schema on top of the silver table:

Dimensions

• DimCustomers — deduplicated customer attributes, surrogate key DimCustomerKey
• DimProducts — deduplicated product attributes, surrogate key DimProductKey
• DimPayments — deduplicated payment types, surrogate key DimPaymentKey
• DimRegions — deduplicated countries, surrogate key DimRegionKey
• DimSales — order-grain descriptive table, surrogate key DimSaleKey

Fact

• FactSales — grain of one row per order, joining all dimension surrogate keys plus measures quantity and unit_price.

erDiagram
    DimCustomers {
        int DimCustomerKey PK
        int customer_id
        string customer_name
        string customer_email
    }
    DimProducts {
        int DimProductKey PK
        int product_id
        string product_name
        string product_category
    }
    DimPayments {
        int DimPaymentKey PK
        string payment_type
    }
    DimRegions {
        int DimRegionKey PK
        string country
    }
    DimSales {
        int DimSaleKey PK
        int order_id
        date order_date
    }
    FactSales {
        int DimSaleKey FK
        int DimCustomerKey FK
        int DimProductKey FK
        int DimRegionKey FK
        int DimPaymentKey FK
        int quantity
        decimal unit_price
    }
    FactSales }o--|| DimSales : references
    FactSales }o--|| DimCustomers : references
    FactSales }o--|| DimProducts : references
    FactSales }o--|| DimRegions : references
    FactSales }o--|| DimPayments : references

Loading

Note: DimSales currently carries several descriptive columns (customer/product/payment details) that are already normalized into their own dimensions. In a stricter Kimball design you'd trim it down to just the order-level attributes (order_id, order_date, last_updated, processDate) since the rest is redundant with the other dimensions — a good next iteration.

---

Slowly Changing Dimensions (SCDs.py)

This notebook is a standalone conceptual demo (separate source/target tables from the main pipeline) showing both SCD strategies on a simple product dataset.

SCD Type 1 — Overwrite History

• Source: datamodeling.default.scdtyp1_source
• Target: datamodeling.gold.scdtyp1_table
• A single MERGE INTO updates matched rows in place (UPDATE SET *) and inserts new ones. When prod_cat for prod_id = 3 is updated, the old value is simply overwritten — no history is kept.

SCD Type 2 — Preserve History

• Source: datamodeling.default.scdtyp2_source

• Target: datamodeling.gold.scdtype2_table (adds start_date, end_date, is_current)

• Implemented as a two-step merge, which is the standard pattern in Spark SQL since a single MERGE can't both expire an old row and insert a new one for the same key:

  1. Merge 1 (Expire): Matches on prod_id + is_current = 'Y'; if any tracked column changed, sets end_date = current_date() and is_current = 'N' on the old row.
  2. Merge 2 (Insert): Matches on the same key/flag; any row from source that is not matched (i.e. brand-new or just-expired) gets inserted fresh with is_current = 'Y' and end_date = '3000-01-01'.

  This preserves the full change history — each version of a product row is retained with its own validity window.

---

How to Run

1. Import all five .py files as Databricks notebooks (they use the # Databricks notebook source / # COMMAND ---------- cell markers).
2. Ensure a Unity Catalog catalog/schema structure exists: datamodeling.default, datamodeling.bronze, datamodeling.silver, datamodeling.gold.
3. Run in order for the main pipeline:
   1. Source.py
   2. Bronze.py
   3. Silver.py
   4. Gold.py
4. Run SCDs.py independently to explore the SCD Type 1 vs Type 2 patterns.
5. Re-run Source.py's incremental insert cell, then re-run Bronze.py → Silver.py → Gold.py to see incremental loading and upserts in action.

---

Key Concepts Demonstrated

• ✅ Medallion Architecture (Bronze / Silver / Gold)
• ✅ Incremental/watermark-based extraction
• ✅ Upserts with MERGE INTO
• ✅ Surrogate key generation with ROW_NUMBER()
• ✅ Star schema design (Fact & Dimension tables)
• ✅ SCD Type 1 (overwrite)
• ✅ SCD Type 2 (historical tracking via two-step merge)

Possible Next Steps

• Make Bronze append-only (currently uses CREATE OR REPLACE) so historical bronze snapshots aren't lost on each run.
• Wire the SCDs.py Type 2 pattern into DimProducts in the main Gold pipeline, so product history is actually tracked end-to-end.
• Parameterize catalog/schema names instead of hardcoding datamodeling.*.
• Add data quality checks (nulls, duplicate PKs) between layers.