Build a Banking Transaction Lakehouse

Build a medallion architecture pipeline for banking data in Databricks.

Introduction

30 Second Summary

Every time you tap your card or receive a salary deposit, a record is created somewhere. Banks process millions of these transactions daily and need a system to transform raw data into clean, trustworthy information.

In this project, you will build a complete data lakehouse pipeline that processes banking transactions through three layers of refinement. You will use Databricks, PySpark, and Delta Lake to go from messy raw data to polished business-ready summaries.

What You'll Build

You will build a medallion architecture pipeline (Bronze, Silver, and Gold layers) in Databricks that ingests raw banking transactions, cleans and validates them, and produces business-ready aggregations stored as Delta Lake tables.

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

  • A medallion architecture pipeline with Bronze, Silver, and Gold layers processing banking transaction data in Databricks.
  • Data quality checks that quarantine bad records and produce quality metrics, plus incremental upserts using MERGE INTO.
  • A star schema dimensional model with fact and dimension tables for business analytics, plus Delta Lake time travel for auditing historical data.
  • Secret Mission: Implement SCD Type 2 to track customer address history over time.

Are there any prerequisites?

No. This project runs entirely in your browser using Databricks Free Edition. You do not need to install any local tools or have prior Databricks experience.

Before We Start

Before We Start...

Set Up Databricks Free Edition

Before you can build a data pipeline, you need a place to run it. Databricks provides a browser-based workspace where you can write and execute PySpark code against Delta Lake tables without installing anything locally.

In this step, you will create a free Databricks account, set up your notebook, and prepare the database schemas that will hold each layer of your pipeline.

In this step, get ready to:
  • Sign up for Databricks Free Edition and open your workspace.
  • Create a notebook and set up your Bronze, Silver, and Gold schemas.
  • Verify that PySpark and Delta Lake are working correctly.
Sign up for Databricks Free Edition

Databricks Free Edition is a permanently free, serverless workspace designed for learning and experimentation. No credit card or cloud account is required.

  • Open your browser and navigate to the Databricks signup page:
  • https://www.databricks.com/try-databricks
  • Under the FOR PERSONAL USE section, click Get Free Edition.
  • Sign up using your personal email address. You can use Google sign-in, Microsoft sign-in, or a standard email with a verification code.
  • If you choose email sign-up, check your inbox for a verification code and enter it on the Databricks page.
  • Wait about 1-2 minutes for your workspace to be created automatically.

What is Databricks Free Edition?

Free Edition gives you a serverless workspace with Python and SQL notebook support, Delta Lake tables, and Unity Catalog. It replaced the legacy Community Edition (retired in 2025) and has no expiration date.

It does have daily compute quotas, but this project is light enough to complete well within those limits.

Once your workspace loads, you should see the Databricks home screen with a left sidebar containing options like +New, Workspace, and Catalog.

Having trouble signing up?

If you encounter an error during signup, try opening an incognito or private browser window and starting the process fresh from the signup page. Temporarily disable any VPN or ad-blocker if the issue persists.

If you still see errors, contact the Free Edition team at free_edition_help@databricks.com with your signup email and a screenshot of the error.

help me troubleshoot my Databricks Free Edition signup error

Create your catalog schemas

Your pipeline will organize data into three layers: Bronze (raw), Silver (cleaned), and Gold (business-ready). Each layer needs its own schema in the Unity Catalog to keep tables separated and organized.

  • In the left sidebar of your workspace, click +New.
  • Select Notebook from the menu that appears.
  • Name the notebook Banking Lakehouse Pipeline.
  • Confirm the default language is set to Python.

Why Python over SQL?

Databricks Free Edition supports both Python and SQL in notebooks. This project uses PySpark (Python) because enterprise data engineering roles explicitly require PySpark skills for transformations, and it gives you more control over complex logic like data quality checks and SCD patterns.

Your notebook opens with an empty first cell. The medallion architecture pattern organizes your lakehouse into three schemas that map to Bronze, Silver, and Gold layers.

  • Create your three schemas by pasting this code into the first cell and clicking the Run button (or pressing Shift+Enter):
spark.sql("CREATE SCHEMA IF NOT EXISTS bronze")
spark.sql("CREATE SCHEMA IF NOT EXISTS silver")
spark.sql("CREATE SCHEMA IF NOT EXISTS gold")

print("Schemas created: bronze, silver, gold")

What does this code do?

  • spark.sql(...) runs SQL statements through PySpark's SparkSession, which is pre-configured in every Databricks notebook.
  • CREATE SCHEMA IF NOT EXISTS creates a new schema only if it does not already exist, making the code safe to re-run.
  • Each schema (bronze, silver, gold) will hold the Delta tables for that layer of your pipeline.
  • Verify the schemas were created by clicking Catalog in the left sidebar.
  • Expand your catalog and confirm you see bronze, silver, and gold listed as schemas.

Don't see your schemas in the Catalog?

Make sure the cell finished running successfully. You should see the output Schemas created: bronze, silver, gold below the cell.

If you see a compute error, your daily quota may have reset. Wait a moment and try running the cell again.

help me troubleshoot why my schemas are not appearing in the Databricks Catalog

Verify your environment

Before building your pipeline, confirm that PySpark can write and read Delta Lake tables. This quick test catches any environment issues early.

  • Click the + Code button below your first cell to add a new cell.
  • Test your environment by pasting and running this code:
# Quick test: create a Delta table, read it back, then clean up
test_data = [("hello_delta", 1)]
df_test = spark.createDataFrame(test_data, ["name", "value"])
df_test.write.format("delta").mode("overwrite").saveAsTable("bronze.test_verify")

# Read the table back to confirm Delta Lake works
display(spark.table("bronze.test_verify"))

# Clean up the test table
spark.sql("DROP TABLE IF EXISTS bronze.test_verify")
print("Environment verified: PySpark and Delta Lake are working!")

What does this test do?

  • It creates a tiny DataFrame with one row and writes it as a Delta table to the bronze schema.
  • It reads the table back using display() to confirm Delta Lake read/write operations work.
  • It drops the test table so your schemas stay clean for the real pipeline.
  • Confirm you see a small table with one row (hello_delta, 1) rendered below the cell, followed by the success message.

Seeing an error when running the test?

If you see a timeout or compute error, serverless compute may still be warming up. Wait 30 seconds and run the cell again.

If the error mentions a missing schema, go back to your first cell and re-run the schema creation code.

help me debug my Databricks Delta Lake test cell error

  • Delete this test cell from your notebook now that the verification is complete. Right-click the cell and select Delete, or click the cell menu (three dots) and choose Delete.

✔️ Awesome, I've got everything!

Great. Double check you have your Banking Lakehouse Pipeline notebook open with the schema creation cell saved.

ⓧ I'd like to double check the full code

Here is the full notebook content you should have at this point:

# Databricks notebook source

# COMMAND ----------
# MAGIC %md
# MAGIC # Banking Transaction Lakehouse Pipeline
# MAGIC ## Medallion Architecture: Bronze -> Silver -> Gold

# COMMAND ----------
# MAGIC %md
# MAGIC ## Setup: Create Schemas

# COMMAND ----------

spark.sql("CREATE SCHEMA IF NOT EXISTS bronze")
spark.sql("CREATE SCHEMA IF NOT EXISTS silver")
spark.sql("CREATE SCHEMA IF NOT EXISTS gold")

print("Schemas created: bronze, silver, gold")

Your Databricks workspace is ready and your schemas are in place. Next up, you will build the Bronze layer by generating simulated banking data and ingesting it into Delta tables.

Build the Bronze Layer

Your schemas are ready and waiting for data. Now it's time to build the first layer of your medallion architecture: the Bronze layer.

The Bronze layer captures raw data exactly as it arrives from source systems. No cleaning, no transformations. Think of it as a secure vault that records everything, even messy or duplicate records, so you always have a complete audit trail to fall back on.

In this step, get ready to:
  • Generate simulated banking transaction and customer data with intentional quality issues.
  • Ingest raw data into Bronze Delta Lake tables with metadata columns.
  • Verify the Bronze layer's append-only behavior by running a second ingestion batch.
Generate simulated banking source data

Real banking pipelines ingest data from core banking systems, payment gateways, and customer databases. For this project, you will simulate that source data using PySpark DataFrames.

The simulated data includes intentional quality issues like null values, duplicates, and inconsistent formatting. These issues are typical of raw data in production systems, and you will clean them in the Silver layer later.

  • Hover below your last cell in the notebook and click + Text to add a Markdown cell.
  • Type the following into the Markdown cell:
## Step 2: Bronze Layer - Raw Ingestion
  • Press Shift+Enter to run the cell and render the heading.
  • Hover below the Markdown cell and click + Code to add a new code cell.
  • Import the PySpark functions and types needed for the entire pipeline by pasting this code into the new cell:
from pyspark.sql.functions import (
    col, lit, current_timestamp, when, trim, upper, lower, initcap,
    count, sum as spark_sum, avg, row_number, to_date, date_format,
    monotonically_increasing_id
)
from pyspark.sql.window import Window
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, IntegerType, DateType
)
from datetime import datetime

What does this code do?

This cell imports all the PySpark functions you will use across the Bronze, Silver, and Gold layers.

  • Functions like col, lit, and current_timestamp help you manipulate DataFrame columns.
  • The Window class enables row-numbering and deduplication logic.
  • The datetime module generates a unique batch identifier for each ingestion run.
  • Press Shift+Enter to run the cell.

The cell completes without output. No errors means the imports are loaded and ready to use.

Seeing a ModuleNotFoundError?

All PySpark functions are pre-installed in the Databricks runtime. If you see an import error, confirm your notebook is attached to a running compute cluster by checking the cluster dropdown in the top-right of the notebook toolbar.

help me debug a ModuleNotFoundError in my Databricks notebook

  • Add a + Text Markdown cell below with the content ### Generate Simulated Banking Transactions and run it.
  • Add a + Code cell below the Markdown heading.
  • Create the raw transactions DataFrame by pasting this code:
transactions_data = [
    ("TXN001", "ACC100", 250.00, "debit", "SuperMart", "2024-03-01 09:15:00", "ZAR", "pos"),
    ("TXN002", "ACC101", 1500.00, "credit", "Salary Deposit", "2024-03-01 08:00:00", "ZAR", "eft"),
    ("TXN003", "ACC100", 89.99, "debit", "Netflix", "2024-03-01 12:00:00", "ZAR", "online"),
    ("TXN004", "ACC102", 3200.00, "debit", "Rent Payment", "2024-03-01 07:00:00", "ZAR", "eft"),
    ("TXN005", "ACC101", 45.50, "debit", "Coffee Shop", "2024-03-02 08:30:00", "ZAR", "pos"),
    ("TXN006", "ACC103", 750.00, "credit", "Refund", "2024-03-02 14:00:00", "ZAR", "online"),
    ("TXN007", "ACC100", 120.00, "debit", "Uber", "2024-03-02 18:45:00", "ZAR", "online"),
    ("TXN008", "ACC102", 5000.00, "credit", "Salary Deposit", "2024-03-02 08:00:00", "ZAR", "eft"),
    ("TXN009", "ACC103", -50.00, "debit", "Error Entry", "2024-03-03 10:00:00", "ZAR", "pos"),
    ("TXN010", "ACC101", 200.00, "debit", "Grocery Store", "2024-03-03 11:30:00", "ZAR", "pos"),
    ("TXN003", "ACC100", 89.99, "debit", "Netflix", "2024-03-01 12:00:00", "ZAR", "online"),
    (None, "ACC100", 100.00, "debit", "Unknown", "2024-03-03 15:00:00", "ZAR", "pos"),
    ("TXN012", "ACC104", 0.00, "credit", "Zero Transfer", "2024-03-03 16:00:00", "ZAR", "eft"),
    ("TXN013", "ACC100", 350.00, "DEBIT", "  Pharmacy  ", "2024-03-04 09:00:00", "ZAR", "pos"),
    ("TXN014", "ACC102", 1800.00, "Credit", "Investment Return", "2024-03-04 10:00:00", "ZAR", "online"),
]

transactions_columns = [
    "transaction_id", "account_id", "amount", "transaction_type",
    "merchant", "timestamp", "currency", "channel"
]

df_transactions_raw = spark.createDataFrame(transactions_data, transactions_columns)
display(df_transactions_raw)

What quality issues are hiding in this data?

This dataset has been designed with realistic data quality problems that you will catch in the Silver layer:

  • Row 11 is a duplicate of TXN003 (same transaction recorded twice).
  • Row 12 has a None transaction_id (null primary key).
  • TXN009 has a negative amount (-50.00) which is invalid for a debit.
  • TXN013 has inconsistent casing ("DEBIT") and extra whitespace in the merchant name.
  • Press Shift+Enter to run the cell.

You should see a formatted table displaying all 15 rows of transaction data with columns like transaction_id, account_id, amount, and merchant.

  • Add a + Text Markdown cell below with the content ### Generate Simulated Customer Data and run it.
  • Add a + Code cell below.
  • Create the raw customers DataFrame by pasting this code:
customers_data = [
    ("ACC100", "john smith", "JOHN.SMITH@EMAIL.COM", "0821234567", "10 Main Road", "Cape Town", "2020-01-15"),
    ("ACC101", "sarah jones", "sarah.jones@email.com", "0839876543", "25 Oak Avenue", "Johannesburg", "2019-06-20"),
    ("ACC102", "thabo mokoena", "THABO@WORK.CO.ZA", "0761112233", "8 Church Street", "Pretoria", "2021-03-10"),
    ("ACC103", "lisa van der berg", "lisa.vdb@gmail.com", "0844445566", "15 Beach Road", "Durban", "2022-08-01"),
    ("ACC104", "michael brown", "michael.b@email.com", None, "30 Hill Street", "Bloemfontein", "2023-01-05"),
    ("ACC100", "john smith", "john.smith@email.com", "0821234567", "10 Main Road", "Cape Town", "2020-01-15"),
]

customers_columns = [
    "customer_id", "name", "email", "phone", "address", "city", "account_open_date"
]

df_customers_raw = spark.createDataFrame(customers_data, customers_columns)
display(df_customers_raw)

What quality issues are in the customer data?

  • Row 6 is a duplicate of ACC100 (same customer recorded twice).
  • Names are all lowercase instead of proper case.
  • Emails have inconsistent casing (some uppercase, some lowercase).
  • ACC104 has a None phone number.
  • Press Shift+Enter to run the cell.

You should see a table with 6 rows showing customer records including customer_id, name, email, and city.

Table not displaying?

If you see a text-based output instead of a formatted table, make sure you are using display() and not print(). The display() function is a Databricks-specific command that renders DataFrames as interactive tables.

help me fix my DataFrame display in Databricks

Ingest into Bronze Delta tables

With your raw data created, you can now write it into Delta Lake tables. The Bronze layer stores data in append mode, meaning every ingestion adds new rows without overwriting previous data.

You will also add three metadata columns to each record. These track when data was ingested, which source system it came from, and which batch it belongs to.

  • Add a + Text Markdown cell below with the content ### Ingest into Bronze Delta Tables and run it.
  • Add a + Code cell below.
  • Write the transactions to a Bronze Delta table by pasting this code:
batch_id = datetime.now().strftime("%Y%m%d%H%M%S")

df_transactions_bronze = df_transactions_raw \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_system", lit("core_banking")) \
    .withColumn("_batch_id", lit(batch_id))

df_transactions_bronze.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.raw_transactions")

print(f"Bronze transactions ingested. Batch ID: {batch_id}")

What does this code do?

  • batch_id generates a unique timestamp string (e.g. 20240301093000) so you can identify which ingestion run produced each row.
  • withColumn adds the three metadata columns (_ingested_at, _source_system, _batch_id) to every row.
  • .format("delta").mode("append") writes the data as a Delta table in append-only mode.
  • .saveAsTable("bronze.raw_transactions") registers the table in the bronze schema of your Unity Catalog.
  • Press Shift+Enter to run the cell.

You should see a message like Bronze transactions ingested. Batch ID: 20260625... printed below the cell.

  • Add another + Code cell below.
  • Write the customers to a Bronze Delta table by pasting this code:
df_customers_bronze = df_customers_raw \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_system", lit("core_banking")) \
    .withColumn("_batch_id", lit(batch_id))

df_customers_bronze.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.raw_customers")

print(f"Bronze customers ingested. Batch ID: {batch_id}")

Why the same batch_id for both tables?

Both tables use the same batch_id because they were ingested together in a single pipeline run. In production, this lets you trace exactly which batch introduced a particular record across all your Bronze tables.

  • Press Shift+Enter to run the cell.

You should see Bronze customers ingested. Batch ID: ... printed with the same batch ID as the transactions cell.

Seeing a table already exists error?

If you see an error about bronze.raw_transactions or bronze.raw_customers already existing, you may have run the cell more than once. This is actually fine for append mode. The data simply gets added again. If you want a clean start, run DROP TABLE bronze.raw_transactions in a new cell, then re-run the ingestion.

help me fix a table already exists error in Databricks

Verify Bronze layer properties

A key property of the Bronze layer is that it is append-only. Every time you run an ingestion, new rows are added without touching existing data.

You will prove this by checking the current row count, running the ingestion a second time, and confirming the count doubles.

  • Add a + Text Markdown cell below with the content ### Verify Bronze Layer (Append-Only) and run it.
  • Add a + Code cell below.
  • Add the verification code by pasting:
bronze_txn_count = spark.table("bronze.raw_transactions").count()
bronze_cust_count = spark.table("bronze.raw_customers").count()
print(f"Bronze transactions: {bronze_txn_count} rows")
print(f"Bronze customers: {bronze_cust_count} rows")
  • Press Shift+Enter to run the cell.

You should see Bronze transactions: 15 rows and Bronze customers: 6 rows from the first ingestion batch.

What does this code do?

spark.table() reads a registered Delta table by its full name (schema.table). The .count() method returns the total number of rows, confirming how much data is stored.

Now simulate a second batch arriving from the source system.

  • Scroll back up to the cell that writes transactions to bronze.raw_transactions (the cell starting with batch_id = datetime.now()...).
  • Click inside that cell and press Shift+Enter to re-run it.
  • Click inside the customers ingestion cell (starting with df_customers_bronze) and press Shift+Enter to re-run it.
  • Scroll back down to your verification cell and press Shift+Enter to re-run it.

You should now see Bronze transactions: 30 rows and Bronze customers: 12 rows. The counts doubled because append mode added the same 15 transactions and 6 customers again with a new _batch_id.

Why is append-only important for Bronze?

In production banking systems, raw data must never be overwritten. Regulators and auditors require a complete record of every piece of data that entered the system.

Append-only ingestion combined with Delta Lake's ACID transactions guarantees that no data is lost, even if the same record arrives multiple times. Deduplication happens in the Silver layer instead.

Row count didn't double?

Make sure you re-ran both the transactions ingestion cell and the customers ingestion cell before re-running the verification cell. Each cell must be run individually by clicking inside it and pressing Shift+Enter.

help me understand why my Bronze table row count is wrong

✔️ Awesome, I've got everything!

Great work. Double check you have all the cells in order and your verification shows 30 transaction rows and 12 customer rows.

ⓧ I'd like to double check the full code

Here is the complete notebook up to this point. Each # COMMAND ---------- separator represents a new cell in your notebook.

# Databricks notebook source

# COMMAND ----------
# MAGIC %md
# MAGIC # Banking Transaction Lakehouse Pipeline
# MAGIC ## Medallion Architecture: Bronze -> Silver -> Gold

# COMMAND ----------
# MAGIC %md
# MAGIC ## Setup: Create Schemas

# COMMAND ----------

spark.sql("CREATE SCHEMA IF NOT EXISTS bronze")
spark.sql("CREATE SCHEMA IF NOT EXISTS silver")
spark.sql("CREATE SCHEMA IF NOT EXISTS gold")

print("Schemas created: bronze, silver, gold")

# COMMAND ----------
# MAGIC %md
# MAGIC ## Step 2: Bronze Layer - Raw Ingestion

# COMMAND ----------

from pyspark.sql.functions import (
    col, lit, current_timestamp, when, trim, upper, lower, initcap,
    count, sum as spark_sum, avg, row_number, to_date, date_format,
    monotonically_increasing_id
)
from pyspark.sql.window import Window
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, IntegerType, DateType
)
from datetime import datetime

# COMMAND ----------
# MAGIC %md
# MAGIC ### Generate Simulated Banking Transactions

# COMMAND ----------

transactions_data = [
    ("TXN001", "ACC100", 250.00, "debit", "SuperMart", "2024-03-01 09:15:00", "ZAR", "pos"),
    ("TXN002", "ACC101", 1500.00, "credit", "Salary Deposit", "2024-03-01 08:00:00", "ZAR", "eft"),
    ("TXN003", "ACC100", 89.99, "debit", "Netflix", "2024-03-01 12:00:00", "ZAR", "online"),
    ("TXN004", "ACC102", 3200.00, "debit", "Rent Payment", "2024-03-01 07:00:00", "ZAR", "eft"),
    ("TXN005", "ACC101", 45.50, "debit", "Coffee Shop", "2024-03-02 08:30:00", "ZAR", "pos"),
    ("TXN006", "ACC103", 750.00, "credit", "Refund", "2024-03-02 14:00:00", "ZAR", "online"),
    ("TXN007", "ACC100", 120.00, "debit", "Uber", "2024-03-02 18:45:00", "ZAR", "online"),
    ("TXN008", "ACC102", 5000.00, "credit", "Salary Deposit", "2024-03-02 08:00:00", "ZAR", "eft"),
    ("TXN009", "ACC103", -50.00, "debit", "Error Entry", "2024-03-03 10:00:00", "ZAR", "pos"),
    ("TXN010", "ACC101", 200.00, "debit", "Grocery Store", "2024-03-03 11:30:00", "ZAR", "pos"),
    ("TXN003", "ACC100", 89.99, "debit", "Netflix", "2024-03-01 12:00:00", "ZAR", "online"),
    (None, "ACC100", 100.00, "debit", "Unknown", "2024-03-03 15:00:00", "ZAR", "pos"),
    ("TXN012", "ACC104", 0.00, "credit", "Zero Transfer", "2024-03-03 16:00:00", "ZAR", "eft"),
    ("TXN013", "ACC100", 350.00, "DEBIT", "  Pharmacy  ", "2024-03-04 09:00:00", "ZAR", "pos"),
    ("TXN014", "ACC102", 1800.00, "Credit", "Investment Return", "2024-03-04 10:00:00", "ZAR", "online"),
]

transactions_columns = [
    "transaction_id", "account_id", "amount", "transaction_type",
    "merchant", "timestamp", "currency", "channel"
]

df_transactions_raw = spark.createDataFrame(transactions_data, transactions_columns)
display(df_transactions_raw)

# COMMAND ----------
# MAGIC %md
# MAGIC ### Generate Simulated Customer Data

# COMMAND ----------

customers_data = [
    ("ACC100", "john smith", "JOHN.SMITH@EMAIL.COM", "0821234567", "10 Main Road", "Cape Town", "2020-01-15"),
    ("ACC101", "sarah jones", "sarah.jones@email.com", "0839876543", "25 Oak Avenue", "Johannesburg", "2019-06-20"),
    ("ACC102", "thabo mokoena", "THABO@WORK.CO.ZA", "0761112233", "8 Church Street", "Pretoria", "2021-03-10"),
    ("ACC103", "lisa van der berg", "lisa.vdb@gmail.com", "0844445566", "15 Beach Road", "Durban", "2022-08-01"),
    ("ACC104", "michael brown", "michael.b@email.com", None, "30 Hill Street", "Bloemfontein", "2023-01-05"),
    ("ACC100", "john smith", "john.smith@email.com", "0821234567", "10 Main Road", "Cape Town", "2020-01-15"),
]

customers_columns = [
    "customer_id", "name", "email", "phone", "address", "city", "account_open_date"
]

df_customers_raw = spark.createDataFrame(customers_data, customers_columns)
display(df_customers_raw)

# COMMAND ----------
# MAGIC %md
# MAGIC ### Ingest into Bronze Delta Tables

# COMMAND ----------

batch_id = datetime.now().strftime("%Y%m%d%H%M%S")

df_transactions_bronze = df_transactions_raw \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_system", lit("core_banking")) \
    .withColumn("_batch_id", lit(batch_id))

df_transactions_bronze.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.raw_transactions")

print(f"Bronze transactions ingested. Batch ID: {batch_id}")

# COMMAND ----------

df_customers_bronze = df_customers_raw \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_system", lit("core_banking")) \
    .withColumn("_batch_id", lit(batch_id))

df_customers_bronze.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.raw_customers")

print(f"Bronze customers ingested. Batch ID: {batch_id}")

# COMMAND ----------
# MAGIC %md
# MAGIC ### Verify Bronze Layer (Append-Only)

# COMMAND ----------

bronze_txn_count = spark.table("bronze.raw_transactions").count()
bronze_cust_count = spark.table("bronze.raw_customers").count()
print(f"Bronze transactions: {bronze_txn_count} rows")
print(f"Bronze customers: {bronze_cust_count} rows")

Your Bronze layer is complete. You have raw banking data safely stored in Delta tables with full metadata tracking and proven append-only behavior. Next up, you will build the Silver layer to clean, validate, and deduplicate this messy data into something trustworthy.

Build the Silver Layer

Your Bronze layer now holds every raw record exactly as it arrived from the source system. The problem is that raw data contains duplicates, null values, negative amounts, and inconsistent formatting.

The Silver layer is where you clean, validate, and deduplicate that data. Records that fail quality checks get quarantined rather than dropped silently, so you always have an audit trail of what was rejected and why.

In this step, get ready to:
  • Implement data quality checks and quarantine bad records.
  • Transform and clean transaction data using MERGE INTO.
  • Transform and clean customer data using MERGE INTO.
Implement data quality checks

Before any transformation happens, you need to identify which records are safe to promote and which are not. In banking data, a transaction with a null ID or a negative amount is untrustworthy.

  • Hover below your last cell and click + Text to add a Markdown cell.
  • Paste the following into the Markdown cell to create a section header:
## Step 3: Silver Layer - Clean and Validate
  • Hover below that Markdown cell and click + Text again to add a second Markdown cell.
  • Paste this subsection header:
### Data Quality Checks
  • Hover below the Markdown cell and click + Code to add a new code cell.
  • Paste the following code that identifies bad records, quarantines them, and prints a quality report:
df_bronze_txn = spark.table("bronze.raw_transactions")

null_ids = df_bronze_txn.filter(col("transaction_id").isNull())
negative_amounts = df_bronze_txn.filter(col("amount") < 0)
zero_credits = df_bronze_txn.filter(
    (col("transaction_type").ilike("credit")) & (col("amount") == 0)
)

window_dedup = Window.partitionBy("transaction_id").orderBy(col("_ingested_at").desc())
df_with_row_num = df_bronze_txn.filter(col("transaction_id").isNotNull()) \
    .withColumn("_row_num", row_number().over(window_dedup))
duplicates = df_with_row_num.filter(col("_row_num") > 1).drop("_row_num")

df_quarantine = null_ids.withColumn("_rejection_reason", lit("null_transaction_id")) \
    .unionByName(
        negative_amounts.withColumn("_rejection_reason", lit("negative_amount")),
        allowMissingColumns=True
    ) \
    .unionByName(
        zero_credits.withColumn("_rejection_reason", lit("zero_credit_amount")),
        allowMissingColumns=True
    )

df_quarantine.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("silver.quarantine_transactions")

total_records = df_bronze_txn.count()
quarantined = df_quarantine.count()
passed = total_records - quarantined

print("=== DATA QUALITY REPORT ===")
print(f"Total records:      {total_records}")
print(f"Passed:             {passed}")
print(f"Quarantined:        {quarantined}")
print(f"Pass rate:          {(passed/total_records)*100:.1f}%")
  • Run the cell by pressing Shift+Enter.

What does this code do?

The code reads all raw transactions from Bronze and applies three quality filters. Records with a null transaction_id, a negative amount, or a zero-amount credit are flagged as bad data.

It also uses a Window function to detect duplicate transaction_id values. All bad records are tagged with a _rejection_reason and written to silver.quarantine_transactions so nothing is silently lost.

You should see a DATA QUALITY REPORT printed below the cell showing total records, passed, quarantined counts, and a pass rate percentage.

Not seeing the quality report?

Make sure all the PySpark imports from earlier cells have already been run. If you see a NameError for col or Window, scroll up and re-run the imports cell first.

If you get a table-not-found error for bronze.raw_transactions, re-run the Bronze ingestion cells above.

help me debug my data quality checks cell

Transform and clean transactions

Now that bad records are quarantined, you can transform the good records into a clean, standardized format. The MERGE INTO pattern ensures that if you re-run this cell, existing records get updated rather than duplicated.

  • Hover below your last cell and click + Text to add a Markdown cell with this header:
### Transform and Load Silver Transactions
  • Click + Code below the Markdown cell, then paste the following transformation and MERGE code:
df_clean_txn = df_with_row_num.filter(col("_row_num") == 1).drop("_row_num")

df_clean_txn = df_clean_txn \
    .filter(col("transaction_id").isNotNull()) \
    .filter(col("amount") > 0) \
    .withColumn("transaction_type", upper(trim(col("transaction_type")))) \
    .withColumn("merchant", trim(col("merchant"))) \
    .withColumn("amount", col("amount").cast("decimal(12,2)")) \
    .withColumn("transaction_timestamp", col("timestamp").cast("timestamp")) \
    .withColumn("transaction_date", to_date(col("timestamp"))) \
    .withColumn("processed_at", current_timestamp()) \
    .select(
        "transaction_id", "account_id", "amount", "transaction_type",
        "merchant", "transaction_timestamp", "transaction_date",
        "currency", "channel", "processed_at"
    )

df_clean_txn.createOrReplaceTempView("staged_transactions")

spark.sql("""
    CREATE TABLE IF NOT EXISTS silver.transactions (
        transaction_id STRING,
        account_id STRING,
        amount DECIMAL(12,2),
        transaction_type STRING,
        merchant STRING,
        transaction_timestamp TIMESTAMP,
        transaction_date DATE,
        currency STRING,
        channel STRING,
        processed_at TIMESTAMP
    ) USING DELTA
""")

spark.sql("""
    MERGE INTO silver.transactions AS target
    USING staged_transactions AS source
    ON target.transaction_id = source.transaction_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

silver_txn_count = spark.table("silver.transactions").count()
print(f"Silver transactions: {silver_txn_count} rows (clean, deduplicated)")
  • Run the cell by pressing Shift+Enter.

What does this code do?

First, it keeps only the most recent version of each transaction (row number 1 from the dedup window). Then it filters out records with null IDs or non-positive amounts, standardizes transaction_type to uppercase, trims whitespace from merchant, and casts columns to their proper types.

The cleaned DataFrame is registered as a temp view called staged_transactions. Then a MERGE INTO statement upserts into silver.transactions. If a transaction_id already exists, it gets updated. If it is new, it gets inserted. This makes your pipeline idempotent.

You should see output showing the row count for silver.transactions with a message confirming the records are clean and deduplicated.

Seeing a MERGE error?

If you see an error about staged_transactions not being found, make sure the createOrReplaceTempView line ran successfully in the same cell. Temp views only persist within the current session.

If the cell times out, this may be a daily compute quota issue. Wait until the quota resets and try again.

help me fix my MERGE INTO error

Transform and clean customers

The same pattern applies to customer data. You will deduplicate, standardize names to title case, normalize emails to lowercase, and use MERGE INTO for idempotent loading.

  • Add a + Text Markdown cell below your last cell with this header:
### Transform and Load Silver Customers
  • Click + Code below the Markdown cell, then paste the customer transformation code:
df_bronze_cust = spark.table("bronze.raw_customers")

window_cust_dedup = Window.partitionBy("customer_id").orderBy(col("_ingested_at").desc())
df_clean_cust = df_bronze_cust \
    .withColumn("_row_num", row_number().over(window_cust_dedup)) \
    .filter(col("_row_num") == 1) \
    .drop("_row_num") \
    .withColumn("name", initcap(trim(col("name")))) \
    .withColumn("email", lower(trim(col("email")))) \
    .withColumn("phone", trim(col("phone"))) \
    .withColumn("address", trim(col("address"))) \
    .withColumn("city", initcap(trim(col("city")))) \
    .withColumn("account_open_date", col("account_open_date").cast("date")) \
    .withColumn("processed_at", current_timestamp()) \
    .select(
        "customer_id", "name", "email", "phone",
        "address", "city", "account_open_date", "processed_at"
    )

df_clean_cust.createOrReplaceTempView("staged_customers")

spark.sql("""
    CREATE TABLE IF NOT EXISTS silver.customers (
        customer_id STRING,
        name STRING,
        email STRING,
        phone STRING,
        address STRING,
        city STRING,
        account_open_date DATE,
        processed_at TIMESTAMP
    ) USING DELTA
""")

spark.sql("""
    MERGE INTO silver.customers AS target
    USING staged_customers AS source
    ON target.customer_id = source.customer_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

silver_cust_count = spark.table("silver.customers").count()
print(f"Silver customers: {silver_cust_count} rows (deduplicated, standardized)")
display(spark.table("silver.customers"))
  • Run the cell by pressing Shift+Enter.

What does this code do?

It reads from bronze.raw_customers and keeps only the latest record per customer_id using a window function. Names get converted to title case with initcap(), emails go to lowercase with lower(), and all string fields are trimmed.

The MERGE INTO on silver.customers matches on customer_id. If the customer already exists, their record is updated. If they are new, a fresh row is inserted. The final display() call renders the cleaned table so you can visually confirm the standardization.

You should see a row count message plus a rendered table showing customer names in title case and emails in lowercase.

Customer table showing duplicates?

If silver.customers shows more rows than expected, make sure the window_cust_dedup line uses _ingested_at in the orderBy clause. This column tells PySpark which record is the most recent for each customer_id.

help me debug my Silver customers cell

✔️ Awesome, I've got everything!

Great. Double check you have run all three Silver layer cells successfully before moving on.

ⓧ I'd like to double check the full code

Here are all three Silver layer code cells for comparison. Make sure yours match:

Cell 1: Data Quality Checks

df_bronze_txn = spark.table("bronze.raw_transactions")

null_ids = df_bronze_txn.filter(col("transaction_id").isNull())
negative_amounts = df_bronze_txn.filter(col("amount") < 0)
zero_credits = df_bronze_txn.filter(
    (col("transaction_type").ilike("credit")) & (col("amount") == 0)
)

window_dedup = Window.partitionBy("transaction_id").orderBy(col("_ingested_at").desc())
df_with_row_num = df_bronze_txn.filter(col("transaction_id").isNotNull()) \
    .withColumn("_row_num", row_number().over(window_dedup))
duplicates = df_with_row_num.filter(col("_row_num") > 1).drop("_row_num")

df_quarantine = null_ids.withColumn("_rejection_reason", lit("null_transaction_id")) \
    .unionByName(
        negative_amounts.withColumn("_rejection_reason", lit("negative_amount")),
        allowMissingColumns=True
    ) \
    .unionByName(
        zero_credits.withColumn("_rejection_reason", lit("zero_credit_amount")),
        allowMissingColumns=True
    )

df_quarantine.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("silver.quarantine_transactions")

total_records = df_bronze_txn.count()
quarantined = df_quarantine.count()
passed = total_records - quarantined

print("=== DATA QUALITY REPORT ===")
print(f"Total records:      {total_records}")
print(f"Passed:             {passed}")
print(f"Quarantined:        {quarantined}")
print(f"Pass rate:          {(passed/total_records)*100:.1f}%")

Cell 2: Transform and Load Silver Transactions

df_clean_txn = df_with_row_num.filter(col("_row_num") == 1).drop("_row_num")

df_clean_txn = df_clean_txn \
    .filter(col("transaction_id").isNotNull()) \
    .filter(col("amount") > 0) \
    .withColumn("transaction_type", upper(trim(col("transaction_type")))) \
    .withColumn("merchant", trim(col("merchant"))) \
    .withColumn("amount", col("amount").cast("decimal(12,2)")) \
    .withColumn("transaction_timestamp", col("timestamp").cast("timestamp")) \
    .withColumn("transaction_date", to_date(col("timestamp"))) \
    .withColumn("processed_at", current_timestamp()) \
    .select(
        "transaction_id", "account_id", "amount", "transaction_type",
        "merchant", "transaction_timestamp", "transaction_date",
        "currency", "channel", "processed_at"
    )

df_clean_txn.createOrReplaceTempView("staged_transactions")

spark.sql("""
    CREATE TABLE IF NOT EXISTS silver.transactions (
        transaction_id STRING,
        account_id STRING,
        amount DECIMAL(12,2),
        transaction_type STRING,
        merchant STRING,
        transaction_timestamp TIMESTAMP,
        transaction_date DATE,
        currency STRING,
        channel STRING,
        processed_at TIMESTAMP
    ) USING DELTA
""")

spark.sql("""
    MERGE INTO silver.transactions AS target
    USING staged_transactions AS source
    ON target.transaction_id = source.transaction_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

silver_txn_count = spark.table("silver.transactions").count()
print(f"Silver transactions: {silver_txn_count} rows (clean, deduplicated)")

Cell 3: Transform and Load Silver Customers

df_bronze_cust = spark.table("bronze.raw_customers")

window_cust_dedup = Window.partitionBy("customer_id").orderBy(col("_ingested_at").desc())
df_clean_cust = df_bronze_cust \
    .withColumn("_row_num", row_number().over(window_cust_dedup)) \
    .filter(col("_row_num") == 1) \
    .drop("_row_num") \
    .withColumn("name", initcap(trim(col("name")))) \
    .withColumn("email", lower(trim(col("email")))) \
    .withColumn("phone", trim(col("phone"))) \
    .withColumn("address", trim(col("address"))) \
    .withColumn("city", initcap(trim(col("city")))) \
    .withColumn("account_open_date", col("account_open_date").cast("date")) \
    .withColumn("processed_at", current_timestamp()) \
    .select(
        "customer_id", "name", "email", "phone",
        "address", "city", "account_open_date", "processed_at"
    )

df_clean_cust.createOrReplaceTempView("staged_customers")

spark.sql("""
    CREATE TABLE IF NOT EXISTS silver.customers (
        customer_id STRING,
        name STRING,
        email STRING,
        phone STRING,
        address STRING,
        city STRING,
        account_open_date DATE,
        processed_at TIMESTAMP
    ) USING DELTA
""")

spark.sql("""
    MERGE INTO silver.customers AS target
    USING staged_customers AS source
    ON target.customer_id = source.customer_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

silver_cust_count = spark.table("silver.customers").count()
print(f"Silver customers: {silver_cust_count} rows (deduplicated, standardized)")
display(spark.table("silver.customers"))

Your Silver layer is complete. You have clean, deduplicated, type-safe data with full traceability of rejected records. Next up, you will aggregate this clean data into business-ready Gold tables with daily summaries and a star schema dimensional model.

Build the Gold Layer

Your Silver layer now holds clean, validated, and deduplicated data. The next question is: how do business teams actually use this data?

The Gold layer answers that question by reshaping Silver data into business-ready aggregations. You will build a star schema with fact tables (measurable events) and dimension tables (descriptive context). This is the pattern analysts and dashboards query directly.

In this step, get ready to:
  • Create a daily transaction summary as a fact table.
  • Build a customer dimension table with surrogate keys.
  • Join facts and dimensions into a star schema.
Create a daily transaction summary

A daily transaction summary collapses individual transactions into one row per day. This is a fact table because it stores measurable business events (counts, totals, averages).

  • Click the + button below your last cell to add a new cell.
  • Change the cell type to Markdown using the language dropdown at the top-right of the cell.
  • Paste this markdown heading and run the cell:
## Step 4: Gold Layer - Business Aggregations

### Daily Transaction Summary (Fact Table)
  • Add a new Python cell below the markdown cell.
  • Build the daily transaction summary by pasting and running this code:
df_silver_txn = spark.table("silver.transactions")

df_daily_summary = df_silver_txn \
    .groupBy("transaction_date") \
    .agg(
        count("*").alias("total_transactions"),
        spark_sum("amount").alias("total_amount"),
        avg("amount").alias("avg_amount"),
        spark_sum(when(col("transaction_type") == "DEBIT", 1).otherwise(0)).alias("debit_count"),
        spark_sum(when(col("transaction_type") == "CREDIT", 1).otherwise(0)).alias("credit_count"),
        spark_sum(when(col("transaction_type") == "DEBIT", col("amount")).otherwise(0)).alias("total_debits"),
        spark_sum(when(col("transaction_type") == "CREDIT", col("amount")).otherwise(0)).alias("total_credits")
    ) \
    .orderBy("transaction_date")

df_daily_summary.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.fact_daily_transactions")

print("Gold: Daily transaction summary created")
display(spark.table("gold.fact_daily_transactions"))

What does this code do?

  • spark.table("silver.transactions") loads your clean Silver data into a DataFrame.
  • groupBy("transaction_date") collapses all transactions for the same day into a single row.
  • The when/otherwise pattern counts debits and credits separately by checking the transaction_type value.
  • Writing with mode("overwrite") means each run replaces the table entirely. This is safe for Gold because aggregations should always reflect the latest Silver state.

The cell output displays a table with one row per transaction date, showing totals and breakdowns for debits versus credits.

Seeing zero rows or an error?

  • If you see zero rows, make sure you ran all the Silver layer cells earlier in the notebook. The Gold layer reads from silver.transactions which must be populated first.
  • If you see a AnalysisException, confirm the gold schema exists by checking that the schema-creation cell at the top of the notebook ran successfully.

Help me debug my Gold layer daily summary

Build a dimension table and star schema

A dimension table provides the descriptive context around facts. In banking, the customer dimension tells you who made each transaction. Connecting facts to dimensions through a surrogate key (an auto-generated ID) creates the star schema pattern that powers enterprise analytics.

  • Add a markdown cell with the heading ### Customer Dimension Table and run it.
  • Add a new Python cell below and create the customer dimension by pasting and running this code:
df_silver_cust = spark.table("silver.customers")

window_sk = Window.orderBy("customer_id")
df_dim_customer = df_silver_cust \
    .withColumn("customer_sk", row_number().over(window_sk)) \
    .select("customer_sk", "customer_id", "name", "email", "city", "account_open_date")

df_dim_customer.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.dim_customer")

print("Gold: Customer dimension created")
display(spark.table("gold.dim_customer"))

What is a surrogate key?

A surrogate key is a system-generated integer (1, 2, 3...) that uniquely identifies each row in a dimension. It decouples the dimension from the source system's natural key (customer_id).

Here, row_number().over(window_sk) generates the surrogate key customer_sk by ordering customers alphabetically by their customer_id. Fact tables reference this key instead of the raw ID.

You should see a table with five customers, each assigned a unique integer customer_sk alongside their name, email, city, and open date.

Seeing duplicate customers or missing columns?

  • If you see duplicates, confirm that the Silver customer MERGE cell ran successfully in the previous step. The Gold dimension reads from silver.customers which should already be deduplicated.
  • If customer_sk is missing, ensure the Window import from the earlier imports cell is still in scope. Try re-running the imports cell at the top of the notebook.

Help me fix my customer dimension table

  • Add a markdown cell with the heading ### Customer Transaction Fact Table (Star Schema) and run it.
  • Add a new Python cell below and complete the star schema by pasting and running this code:
df_customer_facts = df_silver_txn \
    .groupBy("account_id") \
    .agg(
        count("*").alias("transaction_count"),
        spark_sum("amount").alias("total_spend"),
        avg("amount").alias("avg_transaction_value"),
        spark_sum(when(col("transaction_type") == "DEBIT", col("amount")).otherwise(0)).alias("total_debits"),
        spark_sum(when(col("transaction_type") == "CREDIT", col("amount")).otherwise(0)).alias("total_credits")
    )

df_dim = spark.table("gold.dim_customer")

df_fact_customer_txn = df_customer_facts \
    .join(df_dim, df_customer_facts["account_id"] == df_dim["customer_id"], "left") \
    .select(
        df_dim["customer_sk"],
        df_customer_facts["account_id"],
        "transaction_count",
        "total_spend",
        "avg_transaction_value",
        "total_debits",
        "total_credits"
    )

df_fact_customer_txn.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.fact_customer_transactions")

print("Gold: Customer transaction facts created (star schema)")
display(spark.table("gold.fact_customer_transactions"))

How does the star schema join work?

  • df_customer_facts aggregates Silver transactions per account: total spend, average value, and debit/credit breakdowns.
  • The left join connects each account to its customer dimension record, bringing in the surrogate key customer_sk.
  • The final select keeps only the columns needed for analytics: the surrogate key links this fact table back to gold.dim_customer for lookups.

You should see a table with one row per customer showing their customer_sk, total spend, transaction count, and debit/credit totals.

Seeing null values in customer_sk?

  • A null customer_sk means the join could not match an account_id in the transactions to a customer_id in the dimension. This is expected for ACC104 if that customer was filtered during Silver processing.
  • If all rows have null keys, confirm the dimension cell above ran successfully and that gold.dim_customer contains data.

Help me debug my star schema join

Why a star schema for banking data?

In production banking platforms, analysts ask questions like "how much did customers in Cape Town spend last month?" A star schema lets them answer this with a simple join between the fact table (spend metrics) and the dimension table (city). No complex multi-table traversals needed.

✔️ Awesome, I've got everything!

Great. Double check that all three Gold tables appear in your Catalog Explorer under the gold schema.

ⓧ I'd like to double check the full code

Here are all the cells you added in this step. Compare them with your notebook:

# Cell 1: Daily Transaction Summary (Fact Table)

df_silver_txn = spark.table("silver.transactions")

df_daily_summary = df_silver_txn \
    .groupBy("transaction_date") \
    .agg(
        count("*").alias("total_transactions"),
        spark_sum("amount").alias("total_amount"),
        avg("amount").alias("avg_amount"),
        spark_sum(when(col("transaction_type") == "DEBIT", 1).otherwise(0)).alias("debit_count"),
        spark_sum(when(col("transaction_type") == "CREDIT", 1).otherwise(0)).alias("credit_count"),
        spark_sum(when(col("transaction_type") == "DEBIT", col("amount")).otherwise(0)).alias("total_debits"),
        spark_sum(when(col("transaction_type") == "CREDIT", col("amount")).otherwise(0)).alias("total_credits")
    ) \
    .orderBy("transaction_date")

df_daily_summary.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.fact_daily_transactions")

print("Gold: Daily transaction summary created")
display(spark.table("gold.fact_daily_transactions"))
# Cell 2: Customer Dimension Table

df_silver_cust = spark.table("silver.customers")

window_sk = Window.orderBy("customer_id")
df_dim_customer = df_silver_cust \
    .withColumn("customer_sk", row_number().over(window_sk)) \
    .select("customer_sk", "customer_id", "name", "email", "city", "account_open_date")

df_dim_customer.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.dim_customer")

print("Gold: Customer dimension created")
display(spark.table("gold.dim_customer"))
# Cell 3: Customer Transaction Fact Table (Star Schema)

df_customer_facts = df_silver_txn \
    .groupBy("account_id") \
    .agg(
        count("*").alias("transaction_count"),
        spark_sum("amount").alias("total_spend"),
        avg("amount").alias("avg_transaction_value"),
        spark_sum(when(col("transaction_type") == "DEBIT", col("amount")).otherwise(0)).alias("total_debits"),
        spark_sum(when(col("transaction_type") == "CREDIT", col("amount")).otherwise(0)).alias("total_credits")
    )

df_dim = spark.table("gold.dim_customer")

df_fact_customer_txn = df_customer_facts \
    .join(df_dim, df_customer_facts["account_id"] == df_dim["customer_id"], "left") \
    .select(
        df_dim["customer_sk"],
        df_customer_facts["account_id"],
        "transaction_count",
        "total_spend",
        "avg_transaction_value",
        "total_debits",
        "total_credits"
    )

df_fact_customer_txn.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.fact_customer_transactions")

print("Gold: Customer transaction facts created (star schema)")
display(spark.table("gold.fact_customer_transactions"))

Your Gold layer is complete. You now have a full medallion pipeline: raw Bronze data flows through validated Silver tables and emerges as business-ready Gold aggregations. Next up, you will add quality gates across all three layers and explore Delta Lake time travel to see how your data evolved.

Add Quality Gates and Time Travel

You have a complete medallion pipeline flowing from Bronze through Silver to Gold. But how do you know the pipeline is healthy?

In production banking systems, pipelines run on schedules without human oversight. Automated quality gates catch problems before bad data reaches downstream consumers. In this step, you will add assertion checks across all three layers and explore Delta Lake time travel to query how your data looked at any point in history.

In this step, get ready to:
  • Write pipeline health assertions that validate data across Bronze, Silver, and Gold layers.
  • Use Delta Lake time travel to query historical versions and view the full audit trail.
Implement layer-level quality assertions

Quality gates are automated checks that verify your pipeline produced correct results. Each assertion tests one invariant that must hold true if the pipeline ran successfully.

  • In your notebook, click the + button below the last code cell to create a new cell.
  • Change the cell language to Markdown using the language dropdown at the top of the cell.
  • Type the following section heading:
## Step 5: Data Quality Gates and Time Travel
  • Create another Markdown cell below it with the sub-heading:
### Pipeline Quality Assertions
  • Create a new code cell below the markdown headings and paste the following pipeline health checks:
checks = []

bronze_count = spark.table("bronze.raw_transactions").count()
checks.append(("Bronze has data", bronze_count > 0, f"{bronze_count} rows"))

silver_nulls = spark.table("silver.transactions").filter(col("transaction_id").isNull()).count()
checks.append(("Silver has no null IDs", silver_nulls == 0, f"{silver_nulls} nulls"))

silver_total = spark.table("silver.transactions").agg(spark_sum("amount")).collect()[0][0]
gold_total = spark.table("gold.fact_daily_transactions").agg(spark_sum("total_amount")).collect()[0][0]
amounts_match = abs(float(silver_total) - float(gold_total)) < 0.01
checks.append(("Gold totals match Silver", amounts_match, f"Silver={silver_total}, Gold={gold_total}"))

print("=== PIPELINE HEALTH REPORT ===")
print("-" * 50)
for check_name, passed, detail in checks:
    status = "PASS" if passed else "FAIL"
    print(f"[{status}] {check_name} ({detail})")
print("-" * 50)
all_passed = all(passed for _, passed, _ in checks)
print(f"Overall: {'ALL CHECKS PASSED' if all_passed else 'SOME CHECKS FAILED'}")

What does this code do?

  • The checks list collects tuples of (name, boolean result, detail string) for each assertion.
  • Assertion 1 verifies Bronze contains data. If the pipeline failed at ingestion, this catches it immediately.
  • Assertion 2 confirms Silver has no null transaction_id values. Nulls should have been quarantined in the previous step.
  • Assertion 3 compares the total amount in Silver against the Gold daily summary. If these drift apart, an aggregation bug exists.
  • The report loop prints PASS or FAIL for each check, giving you a single view of pipeline health.
  • Run the cell by clicking the play button or pressing Shift+Enter.
  • You should see a PIPELINE HEALTH REPORT with all three checks showing [PASS] and the overall status reading ALL CHECKS PASSED.

Seeing a FAIL status?

If Bronze has data fails, scroll up and re-run the Bronze ingestion cells from Step 2. Your tables may have been dropped.

If Gold totals match Silver fails, re-run the Gold daily summary cell from Step 4. The Gold table may have been overwritten with stale data before your latest Silver MERGE.

Help me debug my pipeline health report showing FAIL

Demonstrate Delta Lake time travel

Every write operation to a Delta table creates a new version. Delta Lake keeps a transaction log that lets you query any previous state of the table. This is called time travel.

Why does time travel matter in banking?

Financial regulators require banks to reproduce data exactly as it existed at a specific point in time. Time travel provides a built-in audit trail without building separate versioning infrastructure.

  • Create a new Markdown cell and add the sub-heading:
### Delta Lake Time Travel
  • View the full audit trail of your Silver transactions table by creating a new code cell and pasting:
# View table history
display(spark.sql("DESCRIBE HISTORY silver.transactions"))

What does DESCRIBE HISTORY show?

This returns a table with one row per operation ever performed on silver.transactions. You will see the CREATE TABLE, the MERGE operations, timestamps, and version numbers. Each version number is queryable.

  • Run the cell. You should see a table listing all operations performed on silver.transactions, including version numbers, timestamps, and operation types like CREATE TABLE and MERGE.
  • Now query the very first version of the table to see what it looked like before any MERGE operations. Create a new code cell and paste:
# Query the first version of the table (before any MERGE operations)
try:
    df_version_0 = spark.sql("SELECT * FROM silver.transactions VERSION AS OF 0")
    print(f"Version 0 had {df_version_0.count()} rows")
    display(df_version_0)
except Exception as e:
    print(f"Version 0 query: {e}")

# Current version
df_current = spark.table("silver.transactions")
print(f"Current version has {df_current.count()} rows")

What does this code do?

  • The VERSION AS OF 0 clause queries the table exactly as it was at version 0. This is the state when the table was first created (before any MERGE wrote data).
  • The try/except block handles the case where version 0 was the CREATE TABLE statement (which would contain zero rows).
  • Comparing version 0 row count against the current version demonstrates how data accumulates through each MERGE operation.
  • Run the cell. You should see a row count for version 0 (likely 0 rows since the CREATE TABLE was version 0) and the current version row count showing your clean transactions.

Seeing an error on the VERSION AS OF query?

If you see an error about the version not existing, check your DESCRIBE HISTORY output. Use the lowest version number listed there instead of 0.

The try/except block should catch this gracefully and print the error message rather than crashing the cell.

Help me understand my Delta Lake time travel error

  • Finally, add a Markdown cell at the bottom to mark your pipeline as complete:
### Pipeline Complete

You have built a full medallion architecture pipeline:
- Bronze: Raw ingestion with metadata (append-only)
- Silver: Cleaned, validated, deduplicated (MERGE INTO)
- Gold: Business aggregations and star schema
- Quality gates and time travel audit trail

✔️ Awesome, I've got everything!

Great. Double check you have all the cells saved in your notebook.

ⓧ I'd like to double check the full code

Here is the complete notebook source. Compare it against your cells to make sure nothing is missing.

# Databricks notebook source

# COMMAND ----------
# MAGIC %md
# MAGIC # Banking Transaction Lakehouse Pipeline
# MAGIC ## Medallion Architecture: Bronze -> Silver -> Gold

# COMMAND ----------

spark.sql("CREATE SCHEMA IF NOT EXISTS bronze")
spark.sql("CREATE SCHEMA IF NOT EXISTS silver")
spark.sql("CREATE SCHEMA IF NOT EXISTS gold")

print("Schemas created: bronze, silver, gold")

# COMMAND ----------

from pyspark.sql.functions import (
    col, lit, current_timestamp, when, trim, upper, lower, initcap,
    count, sum as spark_sum, avg, row_number, to_date, date_format,
    monotonically_increasing_id
)
from pyspark.sql.window import Window
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, IntegerType, DateType
)
from datetime import datetime

# COMMAND ----------

transactions_data = [
    ("TXN001", "ACC100", 250.00, "debit", "SuperMart", "2024-03-01 09:15:00", "ZAR", "pos"),
    ("TXN002", "ACC101", 1500.00, "credit", "Salary Deposit", "2024-03-01 08:00:00", "ZAR", "eft"),
    ("TXN003", "ACC100", 89.99, "debit", "Netflix", "2024-03-01 12:00:00", "ZAR", "online"),
    ("TXN004", "ACC102", 3200.00, "debit", "Rent Payment", "2024-03-01 07:00:00", "ZAR", "eft"),
    ("TXN005", "ACC101", 45.50, "debit", "Coffee Shop", "2024-03-02 08:30:00", "ZAR", "pos"),
    ("TXN006", "ACC103", 750.00, "credit", "Refund", "2024-03-02 14:00:00", "ZAR", "online"),
    ("TXN007", "ACC100", 120.00, "debit", "Uber", "2024-03-02 18:45:00", "ZAR", "online"),
    ("TXN008", "ACC102", 5000.00, "credit", "Salary Deposit", "2024-03-02 08:00:00", "ZAR", "eft"),
    ("TXN009", "ACC103", -50.00, "debit", "Error Entry", "2024-03-03 10:00:00", "ZAR", "pos"),
    ("TXN010", "ACC101", 200.00, "debit", "Grocery Store", "2024-03-03 11:30:00", "ZAR", "pos"),
    ("TXN003", "ACC100", 89.99, "debit", "Netflix", "2024-03-01 12:00:00", "ZAR", "online"),
    (None, "ACC100", 100.00, "debit", "Unknown", "2024-03-03 15:00:00", "ZAR", "pos"),
    ("TXN012", "ACC104", 0.00, "credit", "Zero Transfer", "2024-03-03 16:00:00", "ZAR", "eft"),
    ("TXN013", "ACC100", 350.00, "DEBIT", "  Pharmacy  ", "2024-03-04 09:00:00", "ZAR", "pos"),
    ("TXN014", "ACC102", 1800.00, "Credit", "Investment Return", "2024-03-04 10:00:00", "ZAR", "online"),
]

transactions_columns = [
    "transaction_id", "account_id", "amount", "transaction_type",
    "merchant", "timestamp", "currency", "channel"
]

df_transactions_raw = spark.createDataFrame(transactions_data, transactions_columns)
display(df_transactions_raw)

# COMMAND ----------

customers_data = [
    ("ACC100", "john smith", "JOHN.SMITH@EMAIL.COM", "0821234567", "10 Main Road", "Cape Town", "2020-01-15"),
    ("ACC101", "sarah jones", "sarah.jones@email.com", "0839876543", "25 Oak Avenue", "Johannesburg", "2019-06-20"),
    ("ACC102", "thabo mokoena", "THABO@WORK.CO.ZA", "0761112233", "8 Church Street", "Pretoria", "2021-03-10"),
    ("ACC103", "lisa van der berg", "lisa.vdb@gmail.com", "0844445566", "15 Beach Road", "Durban", "2022-08-01"),
    ("ACC104", "michael brown", "michael.b@email.com", None, "30 Hill Street", "Bloemfontein", "2023-01-05"),
    ("ACC100", "john smith", "john.smith@email.com", "0821234567", "10 Main Road", "Cape Town", "2020-01-15"),
]

customers_columns = [
    "customer_id", "name", "email", "phone", "address", "city", "account_open_date"
]

df_customers_raw = spark.createDataFrame(customers_data, customers_columns)
display(df_customers_raw)

# COMMAND ----------

batch_id = datetime.now().strftime("%Y%m%d%H%M%S")

df_transactions_bronze = df_transactions_raw \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_system", lit("core_banking")) \
    .withColumn("_batch_id", lit(batch_id))

df_transactions_bronze.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.raw_transactions")

print(f"Bronze transactions ingested. Batch ID: {batch_id}")

# COMMAND ----------

df_customers_bronze = df_customers_raw \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_system", lit("core_banking")) \
    .withColumn("_batch_id", lit(batch_id))

df_customers_bronze.write \
    .format("delta") \
    .mode("append") \
    .saveAsTable("bronze.raw_customers")

print(f"Bronze customers ingested. Batch ID: {batch_id}")

# COMMAND ----------

bronze_txn_count = spark.table("bronze.raw_transactions").count()
bronze_cust_count = spark.table("bronze.raw_customers").count()
print(f"Bronze transactions: {bronze_txn_count} rows")
print(f"Bronze customers: {bronze_cust_count} rows")

# COMMAND ----------

df_bronze_txn = spark.table("bronze.raw_transactions")

null_ids = df_bronze_txn.filter(col("transaction_id").isNull())
negative_amounts = df_bronze_txn.filter(col("amount") < 0)
zero_credits = df_bronze_txn.filter(
    (col("transaction_type").ilike("credit")) & (col("amount") == 0)
)

window_dedup = Window.partitionBy("transaction_id").orderBy(col("_ingested_at").desc())
df_with_row_num = df_bronze_txn.filter(col("transaction_id").isNotNull()) \
    .withColumn("_row_num", row_number().over(window_dedup))
duplicates = df_with_row_num.filter(col("_row_num") > 1).drop("_row_num")

df_quarantine = null_ids.withColumn("_rejection_reason", lit("null_transaction_id")) \
    .unionByName(
        negative_amounts.withColumn("_rejection_reason", lit("negative_amount")),
        allowMissingColumns=True
    ) \
    .unionByName(
        zero_credits.withColumn("_rejection_reason", lit("zero_credit_amount")),
        allowMissingColumns=True
    )

df_quarantine.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("silver.quarantine_transactions")

total_records = df_bronze_txn.count()
quarantined = df_quarantine.count()
passed = total_records - quarantined

print("=== DATA QUALITY REPORT ===")
print(f"Total records:      {total_records}")
print(f"Passed:             {passed}")
print(f"Quarantined:        {quarantined}")
print(f"Pass rate:          {(passed/total_records)*100:.1f}%")

# COMMAND ----------

df_clean_txn = df_with_row_num.filter(col("_row_num") == 1).drop("_row_num")

df_clean_txn = df_clean_txn \
    .filter(col("transaction_id").isNotNull()) \
    .filter(col("amount") > 0) \
    .withColumn("transaction_type", upper(trim(col("transaction_type")))) \
    .withColumn("merchant", trim(col("merchant"))) \
    .withColumn("amount", col("amount").cast("decimal(12,2)")) \
    .withColumn("transaction_timestamp", col("timestamp").cast("timestamp")) \
    .withColumn("transaction_date", to_date(col("timestamp"))) \
    .withColumn("processed_at", current_timestamp()) \
    .select(
        "transaction_id", "account_id", "amount", "transaction_type",
        "merchant", "transaction_timestamp", "transaction_date",
        "currency", "channel", "processed_at"
    )

df_clean_txn.createOrReplaceTempView("staged_transactions")

spark.sql("""
    CREATE TABLE IF NOT EXISTS silver.transactions (
        transaction_id STRING,
        account_id STRING,
        amount DECIMAL(12,2),
        transaction_type STRING,
        merchant STRING,
        transaction_timestamp TIMESTAMP,
        transaction_date DATE,
        currency STRING,
        channel STRING,
        processed_at TIMESTAMP
    ) USING DELTA
""")

spark.sql("""
    MERGE INTO silver.transactions AS target
    USING staged_transactions AS source
    ON target.transaction_id = source.transaction_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

silver_txn_count = spark.table("silver.transactions").count()
print(f"Silver transactions: {silver_txn_count} rows (clean, deduplicated)")

# COMMAND ----------

df_bronze_cust = spark.table("bronze.raw_customers")

window_cust_dedup = Window.partitionBy("customer_id").orderBy(col("_ingested_at").desc())
df_clean_cust = df_bronze_cust \
    .withColumn("_row_num", row_number().over(window_cust_dedup)) \
    .filter(col("_row_num") == 1) \
    .drop("_row_num") \
    .withColumn("name", initcap(trim(col("name")))) \
    .withColumn("email", lower(trim(col("email")))) \
    .withColumn("phone", trim(col("phone"))) \
    .withColumn("address", trim(col("address"))) \
    .withColumn("city", initcap(trim(col("city")))) \
    .withColumn("account_open_date", col("account_open_date").cast("date")) \
    .withColumn("processed_at", current_timestamp()) \
    .select(
        "customer_id", "name", "email", "phone",
        "address", "city", "account_open_date", "processed_at"
    )

df_clean_cust.createOrReplaceTempView("staged_customers")

spark.sql("""
    CREATE TABLE IF NOT EXISTS silver.customers (
        customer_id STRING,
        name STRING,
        email STRING,
        phone STRING,
        address STRING,
        city STRING,
        account_open_date DATE,
        processed_at TIMESTAMP
    ) USING DELTA
""")

spark.sql("""
    MERGE INTO silver.customers AS target
    USING staged_customers AS source
    ON target.customer_id = source.customer_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

silver_cust_count = spark.table("silver.customers").count()
print(f"Silver customers: {silver_cust_count} rows (deduplicated, standardized)")
display(spark.table("silver.customers"))

# COMMAND ----------
# MAGIC %md
# MAGIC ## Step 4: Gold Layer - Business Aggregations

# COMMAND ----------

df_silver_txn = spark.table("silver.transactions")

df_daily_summary = df_silver_txn \
    .groupBy("transaction_date") \
    .agg(
        count("*").alias("total_transactions"),
        spark_sum("amount").alias("total_amount"),
        avg("amount").alias("avg_amount"),
        spark_sum(when(col("transaction_type") == "DEBIT", 1).otherwise(0)).alias("debit_count"),
        spark_sum(when(col("transaction_type") == "CREDIT", 1).otherwise(0)).alias("credit_count"),
        spark_sum(when(col("transaction_type") == "DEBIT", col("amount")).otherwise(0)).alias("total_debits"),
        spark_sum(when(col("transaction_type") == "CREDIT", col("amount")).otherwise(0)).alias("total_credits")
    ) \
    .orderBy("transaction_date")

df_daily_summary.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.fact_daily_transactions")

print("Gold: Daily transaction summary created")
display(spark.table("gold.fact_daily_transactions"))

# COMMAND ----------

df_silver_cust = spark.table("silver.customers")

window_sk = Window.orderBy("customer_id")
df_dim_customer = df_silver_cust \
    .withColumn("customer_sk", row_number().over(window_sk)) \
    .select("customer_sk", "customer_id", "name", "email", "city", "account_open_date")

df_dim_customer.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.dim_customer")

print("Gold: Customer dimension created")
display(spark.table("gold.dim_customer"))

# COMMAND ----------

df_customer_facts = df_silver_txn \
    .groupBy("account_id") \
    .agg(
        count("*").alias("transaction_count"),
        spark_sum("amount").alias("total_spend"),
        avg("amount").alias("avg_transaction_value"),
        spark_sum(when(col("transaction_type") == "DEBIT", col("amount")).otherwise(0)).alias("total_debits"),
        spark_sum(when(col("transaction_type") == "CREDIT", col("amount")).otherwise(0)).alias("total_credits")
    )

df_dim = spark.table("gold.dim_customer")

df_fact_customer_txn = df_customer_facts \
    .join(df_dim, df_customer_facts["account_id"] == df_dim["customer_id"], "left") \
    .select(
        df_dim["customer_sk"],
        df_customer_facts["account_id"],
        "transaction_count",
        "total_spend",
        "avg_transaction_value",
        "total_debits",
        "total_credits"
    )

df_fact_customer_txn.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("gold.fact_customer_transactions")

print("Gold: Customer transaction facts created (star schema)")
display(spark.table("gold.fact_customer_transactions"))

# COMMAND ----------
# MAGIC %md
# MAGIC ## Step 5: Data Quality Gates and Time Travel

# COMMAND ----------
# MAGIC %md
# MAGIC ### Pipeline Quality Assertions

# COMMAND ----------

checks = []

bronze_count = spark.table("bronze.raw_transactions").count()
checks.append(("Bronze has data", bronze_count > 0, f"{bronze_count} rows"))

silver_nulls = spark.table("silver.transactions").filter(col("transaction_id").isNull()).count()
checks.append(("Silver has no null IDs", silver_nulls == 0, f"{silver_nulls} nulls"))

silver_total = spark.table("silver.transactions").agg(spark_sum("amount")).collect()[0][0]
gold_total = spark.table("gold.fact_daily_transactions").agg(spark_sum("total_amount")).collect()[0][0]
amounts_match = abs(float(silver_total) - float(gold_total)) < 0.01
checks.append(("Gold totals match Silver", amounts_match, f"Silver={silver_total}, Gold={gold_total}"))

print("=== PIPELINE HEALTH REPORT ===")
print("-" * 50)
for check_name, passed, detail in checks:
    status = "PASS" if passed else "FAIL"
    print(f"[{status}] {check_name} ({detail})")
print("-" * 50)
all_passed = all(passed for _, passed, _ in checks)
print(f"Overall: {'ALL CHECKS PASSED' if all_passed else 'SOME CHECKS FAILED'}")

# COMMAND ----------
# MAGIC %md
# MAGIC ### Delta Lake Time Travel

# COMMAND ----------

# View table history
display(spark.sql("DESCRIBE HISTORY silver.transactions"))

# COMMAND ----------

# Query the first version of the table (before any MERGE operations)
try:
    df_version_0 = spark.sql("SELECT * FROM silver.transactions VERSION AS OF 0")
    print(f"Version 0 had {df_version_0.count()} rows")
    display(df_version_0)
except Exception as e:
    print(f"Version 0 query: {e}")

# Current version
df_current = spark.table("silver.transactions")
print(f"Current version has {df_current.count()} rows")

# COMMAND ----------
# MAGIC %md
# MAGIC ### Pipeline Complete
# MAGIC
# MAGIC You have built a full medallion architecture pipeline:
# MAGIC - Bronze: Raw ingestion with metadata (append-only)
# MAGIC - Silver: Cleaned, validated, deduplicated (MERGE INTO)
# MAGIC - Gold: Business aggregations and star schema
# MAGIC - Quality gates and time travel audit trail

Secret mission

Implement SCD Type 2 for Customer Dimension

Your Gold layer's dim_customer table overwrites old data whenever a customer moves. In production banking, auditors need to see where a customer lived at the time of each transaction. Can you implement SCD Type 2 to preserve full address history while keeping only one record marked as current?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely in Databricks Free Edition, so there are no ongoing costs regardless of what you choose.

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

Resources you used:

  • Databricks Free Edition account (permanently free, no charges).
  • Delta tables stored in your workspace across bronze, silver, and gold schemas.
  • Notebooks: Banking Lakehouse Pipeline and SCD Type 2 Implementation.

Keep everything running

No action needed. Choose this if you plan to keep exploring Databricks or want to extend the pipeline further.

  • Your notebooks and Delta tables persist in your Free Edition workspace indefinitely.
  • Databricks may delete Free Edition accounts that are inactive for a prolonged period, so log in occasionally if you want to keep your work.
  • There are no charges. Databricks Free Edition is permanently free.

Pause - I'll come back to this later

Since Databricks Free Edition uses serverless compute, there are no running processes to shut down. Your notebooks and tables simply sit idle at no cost.

  • Log back in whenever you want to pick up where you left off.
  • Be aware that Databricks may delete Free Edition accounts that are inactive for a prolonged period. Log in occasionally to keep your workspace active.

Delete - I don't want to use this again

Remove all project resources from your workspace. There are no cost savings from deleting since the account is free, but this cleans up your environment.

  • Drop all three schemas and their Delta tables by running these SQL commands in a new notebook cell:
DROP SCHEMA bronze CASCADE;
DROP SCHEMA silver CASCADE;
DROP SCHEMA gold CASCADE;
  • Delete the Banking Lakehouse Pipeline notebook by right-clicking it in the Workspace sidebar and selecting Move to Trash.
  • Delete the SCD Type 2 Implementation notebook the same way.
  • Your Databricks Free Edition account remains active at no cost. You can leave it as is or revisit Databricks in the future.

Nice Work!

Nice Work!

You just built a complete banking transaction lakehouse using Databricks Free Edition. Your pipeline processes raw data through three governed layers, exactly how enterprise banking platforms handle petabyte-scale data in production.

You've learned how to:

  • Build a complete medallion architecture pipeline (Bronze/Silver/Gold) that ingests raw banking transactions, cleans and deduplicates them with PySpark transformations, and produces business-ready aggregations.
  • Implement Delta Lake MERGE INTO for incremental upsert processing with data quality gates, quarantine logic for bad records, and time travel for full audit history.
  • Design a star schema dimensional model with fact and dimension tables in the Gold layer, ready for downstream analytics and reporting.
  • Secret Mission: Implemented SCD Type 2 to track historical changes in customer dimensions, preserving a full address history with effective dates.

Ready to quiz yourself?