Fraud Detection Lakehouse: Bronze & Silver

Build a banking fraud detection lakehouse with Databricks Lakeflow Pipelines.

Introduction

30 Second Summary

Every time you tap your card or make an online purchase, a data pipeline somewhere is evaluating whether that transaction is legitimate. Banks process millions of these events daily, and the systems that catch fraud depend on clean, well-structured data flowing through multiple layers of transformation.

In this project, you will build a banking fraud detection data lakehouse that ingests simulated card transactions through Bronze and Silver layers using Databricks Lakeflow Declarative Pipelines.

What You'll Build

A multi-layer data pipeline that ingests raw banking transactions, applies data quality rules, enriches records with merchant data, and tracks customer changes over time using Change Data Capture.

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

  • A Lakeflow Declarative Pipeline with Bronze streaming tables ingesting transaction, customer, and merchant data via Auto Loader.
  • Silver layer tables with enriched transactions, stream-static joins, deduplication, and fraud indicator columns.
  • Auto CDC with SCD Type 2 tracking full customer dimension change history through Unity Catalog governance.
  • Secret Mission: Handle schema evolution with Auto Loader and detect late-arriving data using windowed expectations.

Are there any prerequisites?

This project assumes experience with data engineering, Python, and SQL. You will also need a Databricks trial account, which provides $400 in credits over 14 days with no credit card required.

Set Up Unity Catalog and Generate Sample Data

Before you can build a pipeline, you need data to flow through it. In production, banking data lands in cloud storage as raw JSON files from upstream systems.

In this step, you will create the governance structure in Unity Catalog and generate synthetic banking data that simulates real transaction feeds, customer change events, and merchant reference files.

In this step, get ready to:
  • Create the Unity Catalog structure (catalog, schema, volume).
  • Generate synthetic banking transaction and customer data.
  • Verify the uploaded data files.
Create the Unity Catalog structure

Unity Catalog organizes data using a three-level namespace: catalog, schema, and table (or volume). You will create a dedicated catalog for this project, a schema inside it, and a managed Volume to store raw landing files.

What is a Volume?

A Volume is a Unity Catalog object that provides governed access to files in cloud storage. Think of it as a managed folder that your pipelines can read from, with full lineage and access control.

  • Create a new notebook by clicking + New in the left sidebar, then selecting Notebook.
  • Click the notebook title at the top (it defaults to "Untitled") and rename it to data_generator.

What language should the notebook use?

Ensure the notebook language is set to Python. If it defaults to SQL or another language, click the language dropdown at the top of the notebook and select Python.

  • In the first cell, create the Unity Catalog resources by pasting the following code:
# Create catalog and schema
spark.sql("CREATE CATALOG IF NOT EXISTS fraud_detection")
spark.sql("CREATE SCHEMA IF NOT EXISTS fraud_detection.lakehouse")
spark.sql("""
    CREATE VOLUME IF NOT EXISTS fraud_detection.lakehouse.raw_data
""")

What does this code do?

These three SQL statements build the namespace hierarchy for your project.

  • fraud_detection is the top-level catalog that groups all project data.
  • fraud_detection.lakehouse is the schema where your pipeline tables will live.
  • fraud_detection.lakehouse.raw_data is the managed Volume where raw JSON landing files are stored.
  • Run the cell by pressing Shift+Enter.
  • You should see the cell complete without errors. The output area may show a brief status message or remain empty.

Seeing a permissions error?

If you see a message about insufficient privileges to create a catalog, confirm you are using a Databricks trial workspace where you have admin rights. Free Edition workspaces may not allow catalog creation.

help me fix a permissions error when creating a Unity Catalog catalog

Generate and upload synthetic banking data

Now you will populate the Volume with realistic banking data. The generator creates three types of files: transaction batches, customer CDC events, and merchant reference data.

  • Hover below the first cell and click + Code to add a new code cell.
  • Set up the data generation configuration by pasting this code:
import json
import random
from datetime import datetime, timedelta

# Configuration
volume_path = "/Volumes/fraud_detection/lakehouse/raw_data"
random.seed(42)

# Helper data
customer_ids = [f"CUST{str(i).zfill(4)}" for i in range(1, 51)]
merchant_ids = [f"MERCH{str(i).zfill(3)}" for i in range(1, 21)]
countries = ["US", "UK", "DE", "FR", "JP", "BR", "IN", "AU"]
cities = {
    "US": ["New York", "Los Angeles", "Chicago"],
    "UK": ["London", "Manchester"],
    "DE": ["Berlin", "Munich"],
    "FR": ["Paris", "Lyon"],
    "JP": ["Tokyo", "Osaka"],
    "BR": ["Sao Paulo"],
    "IN": ["Mumbai", "Delhi"],
    "AU": ["Sydney", "Melbourne"]
}
channels = ["card", "online", "mobile", "atm"]
transaction_types = ["purchase", "withdrawal", "transfer", "refund"]

What does this code do?

This cell imports libraries and sets up the configuration that the data generators will use.

  • volume_path points to your managed Volume using the standard /Volumes/<catalog>/<schema>/<volume>/ path pattern.
  • random.seed(42) ensures reproducible data across runs.
  • The lists define realistic IDs, locations, and transaction types that the generator will sample from.
  • Run this cell with Shift+Enter. You won't see output yet. The next cells use these variables to generate files.
  • Add another code cell and paste the transaction generator:
# Generate transactions (3 batches simulating daily files)
def generate_transactions(batch_num, count, base_date):
    transactions = []
    for i in range(count):
        txn_id = f"TXN{str(batch_num * 1000 + i).zfill(6)}"
        customer_id = random.choice(customer_ids)
        merchant_id = random.choice(merchant_ids)
        country = random.choice(countries)
        city = random.choice(cities[country])

        # Create some fraud patterns
        is_fraud_pattern = random.random() < 0.08
        if is_fraud_pattern:
            amount = round(random.uniform(5000, 25000), 2)
        else:
            amount = round(random.uniform(5, 2000), 2)

        timestamp = base_date + timedelta(
            hours=random.randint(0, 23),
            minutes=random.randint(0, 59),
            seconds=random.randint(0, 59)
        )

        txn = {
            "transaction_id": txn_id,
            "customer_id": customer_id,
            "merchant_id": merchant_id,
            "amount": amount,
            "currency": "USD",
            "transaction_type": random.choice(transaction_types),
            "channel": random.choice(channels),
            "timestamp": timestamp.isoformat() + "Z",
            "location_country": country,
            "location_city": city,
            "card_last_four": str(random.randint(1000, 9999)),
            "is_international": country != "US"
        }
        transactions.append(txn)
    return transactions

base_date = datetime(2024, 6, 1)
for batch in range(3):
    batch_date = base_date + timedelta(days=batch)
    transactions = generate_transactions(batch, 67, batch_date)
    file_path = f"{volume_path}/transactions/batch_{batch + 1}.json"
    content = "\n".join(json.dumps(t) for t in transactions)
    dbutils.fs.put(file_path, content, overwrite=True)
    print(f"Written {len(transactions)} transactions to batch_{batch + 1}.json")

What does this code do?

This function generates 67 transaction records per batch across 3 daily files (about 200 total). Each transaction includes a customer ID, merchant ID, amount, timestamp, location, and channel.

About 8% of transactions get high amounts (over $5,000) to simulate fraud patterns your pipeline will detect later. The dbutils.fs.put call writes each batch as newline-delimited JSON directly into the Volume.

  • Run the cell with Shift+Enter.
  • You should see three lines of output confirming 67 transactions written to each batch file.

Seeing a file path error?

If you see an error about the path not existing, confirm the first cell (catalog/schema/volume creation) ran successfully. The Volume must exist before you can write files to it.

help me fix a file path error when writing to a Unity Catalog Volume

  • Add another code cell and paste the customer CDC event generator:
# Generate customer CDC events (2 batches)
def generate_customer_cdc(batch_num, base_date):
    operations = []
    if batch_num == 0:
        # Initial load - all APPENDs
        for i in range(20):
            cust_id = f"CUST{str(i + 1).zfill(4)}"
            op = {
                "id": cust_id,
                "firstname": f"First{i+1}",
                "lastname": f"Last{i+1}",
                "email": f"customer{i+1}@bank.com",
                "phone": f"+1555{str(random.randint(1000000, 9999999))}",
                "address": f"{random.randint(1, 999)} Main St, City {i+1}",
                "risk_category": random.choice(["low", "medium", "high"]),
                "account_status": "active",
                "operation": "APPEND",
                "operation_date": (base_date + timedelta(hours=i)).isoformat() + "Z"
            }
            operations.append(op)
    else:
        # Updates and deletes
        for i in range(10):
            cust_id = f"CUST{str(random.randint(1, 20)).zfill(4)}"
            op_type = random.choices(["UPDATE", "DELETE"], weights=[0.8, 0.2])[0]
            op = {
                "id": cust_id,
                "firstname": f"Updated{i+1}" if op_type == "UPDATE" else f"First{i+1}",
                "lastname": f"Last{i+1}",
                "email": f"updated{i+1}@bank.com" if op_type == "UPDATE" else f"customer{i+1}@bank.com",
                "phone": f"+1555{str(random.randint(1000000, 9999999))}",
                "address": f"{random.randint(1, 999)} Updated Ave, City {i+1}",
                "risk_category": random.choice(["low", "medium", "high"]),
                "account_status": "active" if op_type == "UPDATE" else "closed",
                "operation": op_type,
                "operation_date": (base_date + timedelta(days=2, hours=i)).isoformat() + "Z"
            }
            operations.append(op)
    return operations

for batch in range(2):
    events = generate_customer_cdc(batch, base_date)
    file_path = f"{volume_path}/customers/cdc_batch_{batch + 1}.json"
    content = "\n".join(json.dumps(e) for e in events)
    dbutils.fs.put(file_path, content, overwrite=True)
    print(f"Written {len(events)} CDC events to cdc_batch_{batch + 1}.json")

What are CDC events?

Change Data Capture (CDC) records track every change to a source system. Each event carries an operation field (APPEND, UPDATE, or DELETE) so downstream systems know what happened.

Batch 1 simulates the initial customer load (20 APPEND operations). Batch 2 simulates later changes (updates to email/address and a few account closures).

  • Run the cell with Shift+Enter.
  • You should see two lines confirming 20 CDC events in batch 1 and 10 in batch 2.
  • Add a final data generation cell and paste the merchant reference generator:
# Generate merchant reference data (1 file)
categories = ["retail", "grocery", "restaurant", "travel", "electronics", "fuel", "entertainment", "healthcare"]
merchants = []
for i in range(20):
    merchant = {
        "merchant_id": f"MERCH{str(i + 1).zfill(3)}",
        "merchant_name": f"Merchant_{i+1}",
        "category": random.choice(categories),
        "risk_score": round(random.uniform(0.01, 0.95), 2),
        "country": random.choice(["US", "UK", "DE", "FR"]),
        "mcc_code": str(random.randint(1000, 9999))
    }
    merchants.append(merchant)

file_path = f"{volume_path}/merchants/merchants.json"
content = "\n".join(json.dumps(m) for m in merchants)
dbutils.fs.put(file_path, content, overwrite=True)
print(f"Written {len(merchants)} merchants to merchants.json")

Why is merchant data separate?

Merchant data is reference data that changes infrequently. In the pipeline, you will load it as a materialized view (batch) rather than a streaming table. This mirrors how production systems treat slowly changing dimensions differently from high-velocity event streams.

  • Run the cell with Shift+Enter.
  • You should see one line confirming 20 merchants written to merchants.json.
Verify uploaded data

Now confirm the generated files are readable and contain the expected structure.

  • Add one more code cell and paste the verification query:
# Verify files
display(spark.sql(f"SELECT * FROM read_files('{volume_path}/transactions/', format => 'json') LIMIT 5"))

What does this code do?

The read_files SQL function reads raw files directly from a path without needing a table definition. It is a quick way to inspect landing data before a pipeline processes it.

  • Run the cell with Shift+Enter.
  • You should see a table displaying 5 JSON records with fields including transaction_id, customer_id, amount, and timestamp.

Seeing an empty result or error?

Make sure all previous cells ran successfully. If you see no data, re-run the transaction generation cell and confirm the print output shows files were written.

help me debug why read_files returns empty results from my Volume

✔️ Awesome, I've got everything!

Great. Double check you see transaction records in the output table, then move on to the next step.

ⓧ I'd like to double check the full code

Here is the complete data_generator notebook. Each section separated by a comment header corresponds to one cell.

# Databricks notebook source
# MAGIC %md
# MAGIC # Fraud Detection Data Generator
# MAGIC Generates synthetic banking data and writes to Unity Catalog Volume

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

# Create catalog and schema
spark.sql("CREATE CATALOG IF NOT EXISTS fraud_detection")
spark.sql("CREATE SCHEMA IF NOT EXISTS fraud_detection.lakehouse")
spark.sql("""
    CREATE VOLUME IF NOT EXISTS fraud_detection.lakehouse.raw_data
""")

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

import json
import random
from datetime import datetime, timedelta

# Configuration
volume_path = "/Volumes/fraud_detection/lakehouse/raw_data"
random.seed(42)

# Helper data
customer_ids = [f"CUST{str(i).zfill(4)}" for i in range(1, 51)]
merchant_ids = [f"MERCH{str(i).zfill(3)}" for i in range(1, 21)]
countries = ["US", "UK", "DE", "FR", "JP", "BR", "IN", "AU"]
cities = {
    "US": ["New York", "Los Angeles", "Chicago"],
    "UK": ["London", "Manchester"],
    "DE": ["Berlin", "Munich"],
    "FR": ["Paris", "Lyon"],
    "JP": ["Tokyo", "Osaka"],
    "BR": ["Sao Paulo"],
    "IN": ["Mumbai", "Delhi"],
    "AU": ["Sydney", "Melbourne"]
}
channels = ["card", "online", "mobile", "atm"]
transaction_types = ["purchase", "withdrawal", "transfer", "refund"]

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

# Generate transactions (3 batches simulating daily files)
def generate_transactions(batch_num, count, base_date):
    transactions = []
    for i in range(count):
        txn_id = f"TXN{str(batch_num * 1000 + i).zfill(6)}"
        customer_id = random.choice(customer_ids)
        merchant_id = random.choice(merchant_ids)
        country = random.choice(countries)
        city = random.choice(cities[country])

        # Create some fraud patterns
        is_fraud_pattern = random.random() < 0.08
        if is_fraud_pattern:
            amount = round(random.uniform(5000, 25000), 2)
        else:
            amount = round(random.uniform(5, 2000), 2)

        timestamp = base_date + timedelta(
            hours=random.randint(0, 23),
            minutes=random.randint(0, 59),
            seconds=random.randint(0, 59)
        )

        txn = {
            "transaction_id": txn_id,
            "customer_id": customer_id,
            "merchant_id": merchant_id,
            "amount": amount,
            "currency": "USD",
            "transaction_type": random.choice(transaction_types),
            "channel": random.choice(channels),
            "timestamp": timestamp.isoformat() + "Z",
            "location_country": country,
            "location_city": city,
            "card_last_four": str(random.randint(1000, 9999)),
            "is_international": country != "US"
        }
        transactions.append(txn)
    return transactions

base_date = datetime(2024, 6, 1)
for batch in range(3):
    batch_date = base_date + timedelta(days=batch)
    transactions = generate_transactions(batch, 67, batch_date)
    file_path = f"{volume_path}/transactions/batch_{batch + 1}.json"
    content = "\n".join(json.dumps(t) for t in transactions)
    dbutils.fs.put(file_path, content, overwrite=True)
    print(f"Written {len(transactions)} transactions to batch_{batch + 1}.json")

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

# Generate customer CDC events (2 batches)
def generate_customer_cdc(batch_num, base_date):
    operations = []
    if batch_num == 0:
        # Initial load - all APPENDs
        for i in range(20):
            cust_id = f"CUST{str(i + 1).zfill(4)}"
            op = {
                "id": cust_id,
                "firstname": f"First{i+1}",
                "lastname": f"Last{i+1}",
                "email": f"customer{i+1}@bank.com",
                "phone": f"+1555{str(random.randint(1000000, 9999999))}",
                "address": f"{random.randint(1, 999)} Main St, City {i+1}",
                "risk_category": random.choice(["low", "medium", "high"]),
                "account_status": "active",
                "operation": "APPEND",
                "operation_date": (base_date + timedelta(hours=i)).isoformat() + "Z"
            }
            operations.append(op)
    else:
        # Updates and deletes
        for i in range(10):
            cust_id = f"CUST{str(random.randint(1, 20)).zfill(4)}"
            op_type = random.choices(["UPDATE", "DELETE"], weights=[0.8, 0.2])[0]
            op = {
                "id": cust_id,
                "firstname": f"Updated{i+1}" if op_type == "UPDATE" else f"First{i+1}",
                "lastname": f"Last{i+1}",
                "email": f"updated{i+1}@bank.com" if op_type == "UPDATE" else f"customer{i+1}@bank.com",
                "phone": f"+1555{str(random.randint(1000000, 9999999))}",
                "address": f"{random.randint(1, 999)} Updated Ave, City {i+1}",
                "risk_category": random.choice(["low", "medium", "high"]),
                "account_status": "active" if op_type == "UPDATE" else "closed",
                "operation": op_type,
                "operation_date": (base_date + timedelta(days=2, hours=i)).isoformat() + "Z"
            }
            operations.append(op)
    return operations

for batch in range(2):
    events = generate_customer_cdc(batch, base_date)
    file_path = f"{volume_path}/customers/cdc_batch_{batch + 1}.json"
    content = "\n".join(json.dumps(e) for e in events)
    dbutils.fs.put(file_path, content, overwrite=True)
    print(f"Written {len(events)} CDC events to cdc_batch_{batch + 1}.json")

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

# Generate merchant reference data (1 file)
categories = ["retail", "grocery", "restaurant", "travel", "electronics", "fuel", "entertainment", "healthcare"]
merchants = []
for i in range(20):
    merchant = {
        "merchant_id": f"MERCH{str(i + 1).zfill(3)}",
        "merchant_name": f"Merchant_{i+1}",
        "category": random.choice(categories),
        "risk_score": round(random.uniform(0.01, 0.95), 2),
        "country": random.choice(["US", "UK", "DE", "FR"]),
        "mcc_code": str(random.randint(1000, 9999))
    }
    merchants.append(merchant)

file_path = f"{volume_path}/merchants/merchants.json"
content = "\n".join(json.dumps(m) for m in merchants)
dbutils.fs.put(file_path, content, overwrite=True)
print(f"Written {len(merchants)} merchants to merchants.json")

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

# Verify files
display(spark.sql(f"SELECT * FROM read_files('{volume_path}/transactions/', format => 'json') LIMIT 5"))

Your data foundation is ready. Next up, you will build the Bronze ingestion layer that reads these raw files using Auto Loader and writes them into governed streaming tables.

Build the Bronze Ingestion Layer

Your raw data is sitting in the Unity Catalog Volume, but it is just flat JSON files. To make this data queryable, governed, and ready for downstream analytics, you need to ingest it into Delta Lake tables.

In the Medallion Architecture, the Bronze layer is where raw data lands with minimal transformation. You will use Lakeflow Declarative Pipelines to define three Bronze datasets that automatically ingest files as they arrive.

In this step, get ready to:
  • Create a Lakeflow Declarative Pipeline targeting your Unity Catalog schema.
  • Write Bronze ingestion code for transactions, customer CDC events, and merchant reference data.
  • Run the pipeline and verify all three datasets with data quality expectations.
Create a new Lakeflow pipeline

A Lakeflow pipeline is a managed environment that orchestrates your data transformations as a directed acyclic graph. You define what each table should look like, and Databricks handles scheduling, retries, and compute.

  • In your Databricks workspace sidebar, click New.
  • Select ETL Pipeline.
  • The Lakeflow Pipelines Editor opens with a default pipeline name. Click the pipeline name at the top and rename it to fraud_detection_pipeline.
  • To the right of the name, click the catalog and schema selector.
  • Set the catalog to fraud_detection and the target schema to lakehouse.

What does setting a destination do?

When you set the catalog and schema, all tables created by this pipeline are automatically registered under fraud_detection.lakehouse in Unity Catalog. This means they are governed, discoverable, and queryable from any notebook or SQL editor in the workspace.

Now create the Python source file where you will write the Bronze layer code.

  • In the pipeline asset browser on the left, click Add, then select Transformation.
  • In the dialog, enter bronze as the Name.
  • Select Python as the Language.
  • Click Create.
  • If a default my_transformation file exists in the transformations folder, you can leave it empty or delete it. Only files with valid pipeline code are executed.

Why a separate file for each layer?

Lakeflow pipelines support multiple source files in the transformations folder. All .py and .sql files in this folder are automatically included as pipeline sources. Splitting Bronze and Silver into separate files keeps each layer maintainable.

Write the Bronze pipeline code

The Bronze layer ingests three data sources using Auto Loader (for streaming file ingestion) and batch reads (for reference data). Each source becomes a table with data quality expectations.

  • Open bronze.py in the transformations folder and paste the following code for the imports and the transactions streaming table:
from pyspark import pipelines as dp
from pyspark.sql.functions import *

# Volume path for raw data
VOLUME_PATH = "/Volumes/fraud_detection/lakehouse/raw_data"

# --- Bronze: Transactions ---
@dp.table(
    name="transactions_bronze",
    comment="Raw card transactions ingested via Auto Loader from landing zone"
)
@dp.expect_or_drop("valid_transaction_id", "transaction_id IS NOT NULL")
@dp.expect("valid_amount", "amount > 0")
@dp.expect("valid_timestamp", "timestamp IS NOT NULL")
def transactions_bronze():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.inferColumnTypes", "true")
            .load(f"{VOLUME_PATH}/transactions/")
    )

What does this code do?

  • from pyspark import pipelines as dp imports the Lakeflow Spark Declarative Pipelines module. This module is only available inside a pipeline context.
  • @dp.table() declares a streaming table. Because the function returns a spark.readStream call, the pipeline knows to process this incrementally as new files land.
  • The cloudFiles format is Auto Loader. It efficiently discovers and processes new JSON files from the Volume path without rescanning old ones.
  • @dp.expect_or_drop() drops rows that fail the condition. @dp.expect() logs violations without dropping rows, which is useful for monitoring data quality trends.

Next, add the customer CDC ingestion below the transactions code.

  • Below the transactions_bronze function, paste the following code:
# --- Bronze: Customers CDC ---
dp.create_streaming_table(
    name="customers_cdc_bronze",
    comment="Raw customer CDC events ingested via Auto Loader"
)

@dp.append_flow(
    target="customers_cdc_bronze",
    name="customers_cdc_ingest_flow"
)
def customers_cdc_ingest_flow():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.inferColumnTypes", "true")
            .load(f"{VOLUME_PATH}/customers/")
    )

Why use create_streaming_table and append_flow here?

This pattern creates a named streaming table target first with dp.create_streaming_table(), then populates it via an @dp.append_flow function.

You will need this pattern in a later step because the CDC processing engine (dp.create_auto_cdc_flow) requires a named source table. The simpler @dp.table() pattern does not create a referenceable target for downstream CDC flows.

Finally, add the merchant reference data as a batch-loaded materialized view.

  • Below the customers_cdc_ingest_flow function, paste the following code:
# --- Bronze: Merchants (Reference Data) ---
@dp.materialized_view(
    name="merchants_bronze",
    comment="Merchant reference data - batch loaded"
)
@dp.expect_or_drop("valid_merchant_id", "merchant_id IS NOT NULL")
@dp.expect("valid_risk_score", "risk_score BETWEEN 0 AND 1")
def merchants_bronze():
    return (
        spark.read
            .format("json")
            .option("inferSchema", "true")
            .load(f"{VOLUME_PATH}/merchants/")
    )

Why a materialized view instead of a streaming table?

Merchant data is reference data that changes infrequently. A materialized view uses spark.read (batch) instead of spark.readStream. It recomputes from source on each pipeline refresh, which is perfect for small, slowly-changing reference tables.

Streaming tables are reserved for data that arrives continuously and needs incremental processing, like the transactions and CDC events.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before moving on.

ⓧ I'd like to double check the full code

Here is the complete bronze.py file for comparison:

from pyspark import pipelines as dp
from pyspark.sql.functions import *

# Volume path for raw data
VOLUME_PATH = "/Volumes/fraud_detection/lakehouse/raw_data"

# --- Bronze: Transactions ---
@dp.table(
    name="transactions_bronze",
    comment="Raw card transactions ingested via Auto Loader from landing zone"
)
@dp.expect_or_drop("valid_transaction_id", "transaction_id IS NOT NULL")
@dp.expect("valid_amount", "amount > 0")
@dp.expect("valid_timestamp", "timestamp IS NOT NULL")
def transactions_bronze():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.inferColumnTypes", "true")
            .load(f"{VOLUME_PATH}/transactions/")
    )

# --- Bronze: Customers CDC ---
dp.create_streaming_table(
    name="customers_cdc_bronze",
    comment="Raw customer CDC events ingested via Auto Loader"
)

@dp.append_flow(
    target="customers_cdc_bronze",
    name="customers_cdc_ingest_flow"
)
def customers_cdc_ingest_flow():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.inferColumnTypes", "true")
            .load(f"{VOLUME_PATH}/customers/")
    )

# --- Bronze: Merchants (Reference Data) ---
@dp.materialized_view(
    name="merchants_bronze",
    comment="Merchant reference data - batch loaded"
)
@dp.expect_or_drop("valid_merchant_id", "merchant_id IS NOT NULL")
@dp.expect("valid_risk_score", "risk_score BETWEEN 0 AND 1")
def merchants_bronze():
    return (
        spark.read
            .format("json")
            .option("inferSchema", "true")
            .load(f"{VOLUME_PATH}/merchants/")
    )
Run the pipeline to validate Bronze

With all three Bronze datasets defined, you can execute the pipeline and watch it build the DAG.

  • Click Run pipeline in the upper right corner of the Pipelines Editor.
  • Wait for the pipeline run to complete. This may take 1-2 minutes while Databricks provisions serverless compute.

What should I see?

The Pipeline Graph on the right side of the editor displays your three datasets: transactions_bronze, customers_cdc_bronze, and merchants_bronze. Each one shows a green checkmark when it completes successfully.

  • Click on the transactions_bronze node in the Pipeline Graph to inspect its details.
  • Review the expectations results. You should see pass rates for valid_transaction_id, valid_amount, and valid_timestamp.

What are Expectations?

Expectations are data quality rules declared directly on your pipeline tables. They run at ingestion time and give you immediate visibility into data issues.

The expect_or_drop variant silently removes bad records. The expect variant lets records through but logs the violation count. Both show results in the pipeline run details.

Pipeline run failed?

  • Check that the catalog is set to fraud_detection and the schema to lakehouse in the pipeline settings. A mismatch causes permission errors.
  • Verify that your Volume path matches exactly: /Volumes/fraud_detection/lakehouse/raw_data. A typo in the path results in a file-not-found error.
  • Confirm that the data generator notebook ran successfully in the previous step and files exist in the volume subdirectories.

Help me debug my Lakeflow pipeline Bronze layer run failure.

Your Bronze layer is live. Raw data flows from JSON files into governed Delta tables with quality checks at the gate. Next up, you will build the Silver layer to enrich these transactions with merchant data and add fraud detection indicators.

Build the Silver Enrichment Layer

Your Bronze layer now reliably ingests raw data from landing zones into Delta Lake tables. But raw data is noisy.

Transaction records arrive without merchant context, lack useful time-based features, and can contain duplicates. The Silver layer solves this by enriching, transforming, and deduplicating the Bronze data into analysis-ready tables. In this step, you will build a Silver layer that joins transactions with merchant data, derives fraud indicator columns, and removes duplicates.

In this step, get ready to:
  • Create a Silver transformation file with an enriched transactions streaming table.
  • Add a deduplicated materialized view with quality expectations.
  • Run the pipeline and verify the Silver tables contain joined, enriched data.
Create the Silver transformation file

The Silver layer reads from Bronze tables and applies business logic. Your first Silver table will join transactions with merchant reference data and compute derived columns that help downstream fraud detection.

  • Return to the fraud_detection_pipeline in the Lakeflow Pipelines Editor.
  • In the pipeline asset browser on the left, click the add button and select Transformation.
  • Choose Python as the language.
  • Name the file silver and click Create.

Why a separate file?

Lakeflow pipelines treat every .py file in the transformations folder as part of the same pipeline DAG. Separating Bronze and Silver into their own files keeps each layer's logic self-contained while Databricks automatically resolves the dependencies between them.

  • Replace the default contents of silver.py with the following code:
from pyspark import pipelines as dp
from pyspark.sql.functions import *
from pyspark.sql.window import Window

# --- Silver: Enriched Transactions ---
@dp.table(
    name="transactions_enriched",
    comment="Transactions enriched with merchant data and fraud indicators"
)
@dp.expect_or_drop("valid_customer", "customer_id IS NOT NULL")
@dp.expect_or_drop("valid_merchant", "merchant_id IS NOT NULL")
@dp.expect("valid_amount_range", "amount BETWEEN 0.01 AND 100000")
def transactions_enriched():
    transactions = spark.readStream.table("transactions_bronze")
    merchants = spark.read.table("merchants_bronze")

    return (
        transactions
            .join(merchants, "merchant_id", "left")
            .withColumn("transaction_hour", hour(col("timestamp").cast("timestamp")))
            .withColumn("transaction_day_of_week", dayofweek(col("timestamp").cast("timestamp")))
            .withColumn("is_high_value", col("amount") > 5000)
            .withColumn("is_international", col("location_country") != lit("US"))
            .select(
                "transaction_id",
                "customer_id",
                "merchant_id",
                "merchant_name",
                "category",
                "amount",
                "currency",
                "transaction_type",
                "channel",
                "timestamp",
                "location_country",
                "location_city",
                "card_last_four",
                "is_international",
                "transaction_hour",
                "transaction_day_of_week",
                "is_high_value",
                "risk_score"
            )
    )

What does this code do?

This defines a streaming table called transactions_enriched that reads the Bronze transactions as a stream and joins them with the merchants materialized view.

  • spark.readStream.table("transactions_bronze") reads transactions as a stream so new data flows through automatically.
  • spark.read.table("merchants_bronze") reads merchants as a static table. This creates a stream-static join where the streaming side picks up the latest merchant snapshot on each micro-batch.
  • The withColumn calls derive fraud indicator columns: the hour and day of week (fraud patterns cluster at certain times), whether the amount exceeds 5000, and whether the transaction is international.
  • The @dp.expect_or_drop decorators drop rows missing a customer or merchant ID. The @dp.expect decorator logs rows outside the valid amount range without dropping them, so you can monitor data quality trends.
Add deduplication and quality expectations

Duplicate records are common in streaming pipelines. A retry, a late-arriving file re-delivery, or an upstream bug can produce multiple rows for the same transaction. A deduplication layer ensures downstream consumers see exactly one row per transaction.

  • Add the following code below the transactions_enriched function in silver.py:
# --- Silver: Deduplicated Transactions ---
@dp.materialized_view(
    name="transactions_deduped",
    comment="Deduplicated transactions - latest record per transaction_id"
)
@dp.expect("no_duplicates", "row_num = 1")
def transactions_deduped():
    window_spec = Window.partitionBy("transaction_id").orderBy(col("timestamp").desc())

    return (
        spark.read.table("transactions_enriched")
            .withColumn("row_num", row_number().over(window_spec))
            .filter(col("row_num") == 1)
            .drop("row_num")
    )

What does this code do?

This defines a materialized view (batch recomputation) that deduplicates the enriched transactions.

  • Window.partitionBy("transaction_id").orderBy(col("timestamp").desc()) groups rows by transaction ID and ranks them newest-first.
  • row_number() assigns 1 to the most recent record per group. The filter keeps only rank 1, discarding duplicates.
  • The @dp.expect("no_duplicates", "row_num = 1") decorator logs if any row somehow has a rank other than 1 after filtering. This acts as a sanity check rather than a drop rule.

Why a materialized view instead of a streaming table?

Deduplication with ROW_NUMBER() requires reading all records to rank them. A materialized view recomputes from scratch on each pipeline refresh, which is the correct approach for global deduplication across all data.

  • Confirm your silver.py file is saved.

✔️ Awesome, I've got everything!

Great. Double check you have saved silver.py in the transformations folder.

ⓧ I'd like to double check the full code

from pyspark import pipelines as dp
from pyspark.sql.functions import *
from pyspark.sql.window import Window

# --- Silver: Enriched Transactions ---
@dp.table(
    name="transactions_enriched",
    comment="Transactions enriched with merchant data and fraud indicators"
)
@dp.expect_or_drop("valid_customer", "customer_id IS NOT NULL")
@dp.expect_or_drop("valid_merchant", "merchant_id IS NOT NULL")
@dp.expect("valid_amount_range", "amount BETWEEN 0.01 AND 100000")
def transactions_enriched():
    transactions = spark.readStream.table("transactions_bronze")
    merchants = spark.read.table("merchants_bronze")

    return (
        transactions
            .join(merchants, "merchant_id", "left")
            .withColumn("transaction_hour", hour(col("timestamp").cast("timestamp")))
            .withColumn("transaction_day_of_week", dayofweek(col("timestamp").cast("timestamp")))
            .withColumn("is_high_value", col("amount") > 5000)
            .withColumn("is_international", col("location_country") != lit("US"))
            .select(
                "transaction_id",
                "customer_id",
                "merchant_id",
                "merchant_name",
                "category",
                "amount",
                "currency",
                "transaction_type",
                "channel",
                "timestamp",
                "location_country",
                "location_city",
                "card_last_four",
                "is_international",
                "transaction_hour",
                "transaction_day_of_week",
                "is_high_value",
                "risk_score"
            )
    )

# --- Silver: Deduplicated Transactions ---
@dp.materialized_view(
    name="transactions_deduped",
    comment="Deduplicated transactions - latest record per transaction_id"
)
@dp.expect("no_duplicates", "row_num = 1")
def transactions_deduped():
    window_spec = Window.partitionBy("transaction_id").orderBy(col("timestamp").desc())

    return (
        spark.read.table("transactions_enriched")
            .withColumn("row_num", row_number().over(window_spec))
            .filter(col("row_num") == 1)
            .drop("row_num")
    )
Run and inspect Silver results

With both Bronze and Silver transformations in place, the pipeline DAG now traces from raw Volume files through Bronze ingestion into enriched, deduplicated Silver tables. Running the pipeline validates the entire flow end to end.

  • Click Start in the top-right corner of the pipeline editor to run the full pipeline.
  • Wait for the run to complete. The Pipeline Graph should now show five datasets: the three Bronze tables plus transactions_enriched and transactions_deduped.

What should I see?

The Pipeline Graph displays a DAG with arrows flowing from transactions_bronze and merchants_bronze into transactions_enriched, then from transactions_enriched into transactions_deduped. All datasets show green checkmarks. The expectations panel shows pass rates for each quality rule.

  • Verify the enriched table contains merchant data by running this query in a new SQL cell or the SQL editor:
SELECT transaction_id, merchant_name, category, transaction_hour,
       transaction_day_of_week, is_high_value, risk_score
FROM fraud_detection.lakehouse.transactions_enriched
LIMIT 10
  • Confirm that merchant_name, category, and the derived columns (transaction_hour, transaction_day_of_week, is_high_value) are populated in the results.
  • Verify deduplication worked by running:
SELECT 'enriched' AS table_name, COUNT(*) AS row_count
FROM fraud_detection.lakehouse.transactions_enriched
UNION ALL
SELECT 'deduped', COUNT(*)
FROM fraud_detection.lakehouse.transactions_deduped
  • Confirm the deduped row count is equal to or fewer than the enriched row count.

Pipeline run failed?

  • Check that silver.py is inside the transformations folder of your pipeline, not in a separate workspace location.
  • Verify you have the from pyspark.sql.window import Window import at the top. Missing this causes a NameError on the deduplication step.
  • If you see a table-not-found error for transactions_bronze, make sure bronze.py is still in the same transformations folder and the pipeline is configured to target the fraud_detection.lakehouse schema.

Help me debug my Silver pipeline run failure in Databricks Lakeflow.

Your Silver layer is live. Transactions now carry merchant context and fraud indicators, and duplicates are cleaned out. Next up, you will implement Change Data Capture to track how customer records evolve over time.

Implement Auto CDC with SCD Type 2

Your Silver layer now enriches transactions with merchant data and fraud indicators. But there's a second dimension to track: customers themselves change over time.

When a customer updates their address or gets flagged as high-risk, you need both the current state (for real-time decisions) and the full history (for auditing). In this step, you'll implement Change Data Capture using Databricks Auto CDC to maintain both views automatically.

In this step, get ready to:
  • Create a cleaned CDC streaming table with quality expectations.
  • Implement SCD Type 1 (current state) and SCD Type 2 (full history) for customer tracking.
  • Run the pipeline and verify CDC behavior with temporal columns.
Create the Silver CDC file

The customers_cdc_bronze table already holds raw CDC events (APPEND, UPDATE, DELETE operations). Before feeding these into the CDC engine, you need a cleaned intermediate table that enforces data quality.

  • In the pipeline's transformations folder, create a new Python file called silver_cdc.
  • Add the following code to define the cleaned CDC streaming table:
from pyspark import pipelines as dp
from pyspark.sql.functions import *

# --- Silver CDC: Clean customer CDC events ---
dp.create_streaming_table(
    name="customers_cdc_clean",
    comment="Cleaned customer CDC events with quality enforcement",
    expect_all_or_drop={
        "valid_id": "id IS NOT NULL",
        "valid_operation": "operation IN ('APPEND', 'DELETE', 'UPDATE')",
        "no_rescued_data": "_rescued_data IS NULL"
    }
)

@dp.append_flow(
    target="customers_cdc_clean",
    name="customers_cdc_clean_flow"
)
def customers_cdc_clean_flow():
    return (
        spark.readStream.table("customers_cdc_bronze")
            .select(
                "id",
                "firstname",
                "lastname",
                "email",
                "phone",
                "address",
                "risk_category",
                "account_status",
                "operation",
                "operation_date",
                "_rescued_data"
            )
    )

What does this code do?

This uses the same dp.create_streaming_table + @dp.append_flow pattern you saw in Bronze. The target table needs to exist as a named entity before the CDC engine can reference it.

  • expect_all_or_drop is a dictionary of expectations. Any record that fails ANY condition in the dict gets dropped entirely.
  • _rescued_data IS NULL catches schema drift. Auto Loader places malformed fields into _rescued_data, so checking it is NULL confirms the record parsed cleanly.
  • The .select() explicitly names every column, including _rescued_data, so the expectation can evaluate it before the row is committed.

Why this pattern instead of @dp.table()?

The dp.create_streaming_table function creates a named target that dp.create_auto_cdc_flow can reference as its source parameter. The simpler @dp.table() decorator does not produce a referenceable name for downstream CDC processing.

Implement SCD Type 1 and SCD Type 2

With clean CDC events flowing into customers_cdc_clean, you can now create two downstream tables. SCD Type 1 keeps only the latest state per customer (great for real-time lookups). SCD Type 2 preserves the full history with temporal columns (great for auditing).

  • Add the SCD Type 1 (current state) section below your existing code in silver_cdc.py:
# --- Silver CDC: Current Customer State (SCD Type 1) ---
dp.create_streaming_table(
    name="customers_current",
    comment="Current state of each customer - SCD Type 1"
)

dp.create_auto_cdc_flow(
    target="customers_current",
    source="customers_cdc_clean",
    keys=["id"],
    sequence_by=col("operation_date"),
    ignore_null_updates=False,
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "operation_date", "_rescued_data"],
)

What does this code do?

  • dp.create_auto_cdc_flow is the Auto CDC engine. It reads CDC events from customers_cdc_clean and applies them as inserts, updates, or deletes to the target table.
  • keys=["id"] identifies which column uniquely identifies a customer.
  • sequence_by=col("operation_date") tells the engine how to order events when multiple arrive for the same key.
  • apply_as_deletes=expr("operation = 'DELETE'") instructs the engine to remove rows from the target when it encounters a DELETE event.
  • except_column_list excludes CDC metadata columns from the final table. You don't want operation or _rescued_data in the customer dimension table.
  • Now add the SCD Type 2 (full history) section below the SCD Type 1 code:
# --- Silver CDC: Customer History (SCD Type 2) ---
dp.create_streaming_table(
    name="customers_history",
    comment="Full customer change history - SCD Type 2"
)

dp.create_auto_cdc_flow(
    target="customers_history",
    source="customers_cdc_clean",
    keys=["id"],
    sequence_by=col("operation_date"),
    ignore_null_updates=False,
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "operation_date", "_rescued_data"],
    stored_as_scd_type="2",
)

What does this code do?

This is nearly identical to the SCD Type 1 flow, with one key addition: stored_as_scd_type="2".

Instead of overwriting the previous state, this parameter tells the engine to close out the old row (by setting __END_AT) and insert a new row with a fresh __START_AT. This gives you a complete audit trail of every change.

  • Save silver_cdc.py.

Seeing a syntax error?

  • Check that you imported both col and expr via the from pyspark.sql.functions import * line at the top of the file.
  • Make sure stored_as_scd_type="2" uses a string "2", not the integer 2.
  • Verify the trailing comma after each parameter in the dp.create_auto_cdc_flow calls. Missing commas between keyword arguments cause syntax errors.

Help me debug my silver_cdc.py file

✔️ Awesome, I've got everything!

Great. Double check you've saved silver_cdc.py before moving on to run the pipeline.

ⓧ I'd like to double check the full code

Here is the complete silver_cdc.py file:

from pyspark import pipelines as dp
from pyspark.sql.functions import *

# --- Silver CDC: Clean customer CDC events ---
dp.create_streaming_table(
    name="customers_cdc_clean",
    comment="Cleaned customer CDC events with quality enforcement",
    expect_all_or_drop={
        "valid_id": "id IS NOT NULL",
        "valid_operation": "operation IN ('APPEND', 'DELETE', 'UPDATE')",
        "no_rescued_data": "_rescued_data IS NULL"
    }
)

@dp.append_flow(
    target="customers_cdc_clean",
    name="customers_cdc_clean_flow"
)
def customers_cdc_clean_flow():
    return (
        spark.readStream.table("customers_cdc_bronze")
            .select(
                "id",
                "firstname",
                "lastname",
                "email",
                "phone",
                "address",
                "risk_category",
                "account_status",
                "operation",
                "operation_date",
                "_rescued_data"
            )
    )

# --- Silver CDC: Current Customer State (SCD Type 1) ---
dp.create_streaming_table(
    name="customers_current",
    comment="Current state of each customer - SCD Type 1"
)

dp.create_auto_cdc_flow(
    target="customers_current",
    source="customers_cdc_clean",
    keys=["id"],
    sequence_by=col("operation_date"),
    ignore_null_updates=False,
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "operation_date", "_rescued_data"],
)

# --- Silver CDC: Customer History (SCD Type 2) ---
dp.create_streaming_table(
    name="customers_history",
    comment="Full customer change history - SCD Type 2"
)

dp.create_auto_cdc_flow(
    target="customers_history",
    source="customers_cdc_clean",
    keys=["id"],
    sequence_by=col("operation_date"),
    ignore_null_updates=False,
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "operation_date", "_rescued_data"],
    stored_as_scd_type="2",
)
Run and verify CDC behavior
  • Click Run pipeline in the pipeline editor to trigger a full refresh.
  • Wait for the run to complete. The Pipeline Graph should now show the full DAG including customers_cdc_clean, customers_current, and customers_history.

Once the pipeline run succeeds, verify the CDC tables are working correctly.

  • Query customers_current to confirm it shows only the latest state per customer by running this SQL:
SELECT * FROM fraud_detection.lakehouse.customers_current
ORDER BY id
LIMIT 10

What should I see?

You should see one row per active customer. Customers that received a DELETE event will not appear in this table at all. The columns include id, firstname, lastname, and other customer attributes, but NOT the operation or operation_date columns (those were excluded).

  • Query customers_history to see the full change history with temporal columns:
SELECT * FROM fraud_detection.lakehouse.customers_history
WHERE id = 'CUST0001'
ORDER BY __START_AT

What should I see?

You should see multiple rows for the same customer, each with __START_AT and __END_AT timestamp columns. The most recent active row has a NULL __END_AT (still current). Older rows have __END_AT populated, showing when that version was superseded.

  • Verify that DELETE operations populate __END_AT by running this query:
SELECT id, account_status, __START_AT, __END_AT
FROM fraud_detection.lakehouse.customers_history
WHERE __END_AT IS NOT NULL
ORDER BY __END_AT DESC
LIMIT 5

You should see rows where __END_AT is populated. These represent customer records that were either updated (superseded by a new version) or deleted.

Pipeline run failing on the CDC tables?

  • Check the pipeline event log for the specific error. The most common issue is a typo in the source parameter. It must exactly match the table name string "customers_cdc_clean".
  • If you see an error about col or expr not being defined, confirm the import line at the top of your file reads from pyspark.sql.functions import *.
  • Ensure the silver_cdc.py file is saved inside the pipeline's transformations folder, not in a separate notebook.

Help me troubleshoot my CDC pipeline run

Your lakehouse now tracks both the current state and complete history of every customer. In the next step, you'll run the full end-to-end pipeline and explore the governance features that tie everything together.

Validate the Pipeline and Explore Governance

You have a fully built data lakehouse with Bronze ingestion, Silver enrichment, and CDC tracking across three transformation files.

Now it is time to run the complete pipeline end-to-end, explore the governance features that Unity Catalog provides, and validate that your business logic produces correct results.

In this step, get ready to:
  • Run a full pipeline refresh and review the complete DAG.
  • Explore Unity Catalog lineage for your enriched transactions table.
  • Query tables to validate joins, deduplication, and CDC behavior.
Run the complete end-to-end pipeline

You have run the pipeline incrementally as you added each file. A full refresh re-processes all source data through every table in one pass, confirming that all three transformation files work together as a single DAG.

  • Return to the fraud_detection_pipeline in the Lakeflow Pipelines Editor.
  • Confirm that the transformations folder lists all three files: bronze.py, silver.py, and silver_cdc.py.
  • Click Run pipeline to trigger a full refresh.

The pipeline will take a few minutes to process all eight tables.

  • Watch the Pipeline Graph as it renders the complete DAG from Bronze through Silver.
  • Once the run completes, click on the run in the pipeline history to open the event log.
  • Review the event log for any warnings or failures.

What should I see in the Pipeline Graph?

The graph displays all eight datasets as nodes with arrows showing data flow. You should see transactions_bronze feeding into transactions_enriched, which feeds into transactions_deduped.

On the CDC side, customers_cdc_bronze flows into customers_cdc_clean, which then branches to both customers_current and customers_history. Green checkmarks indicate successful processing.

  • Verify that all eight datasets show green checkmarks in the Pipeline Graph.
  • Click on any dataset node to view its expectations results in the run details panel.

Seeing a failed dataset?

Click the failed node in the graph to see the error message. Common causes include a typo in a column name referenced in an expectation or a missing import in one of your transformation files.

If the CDC tables fail, verify that silver_cdc.py references customers_cdc_bronze (not customers_bronze) as the upstream source.

Help me debug my pipeline run failure.

Explore Unity Catalog lineage

Unity Catalog automatically tracks where data comes from and where it flows to. This lineage metadata is invaluable for impact analysis, debugging, and compliance audits.

  • In the left sidebar, click Catalog to open the Catalog Explorer.
  • Expand fraud_detection > lakehouse in the catalog tree.
  • Click on transactions_enriched to open its detail page.
  • Click the Lineage tab.

You should see a lineage graph showing upstream dependencies flowing into transactions_enriched.

What does the lineage graph tell you?

The graph traces the full data journey. Raw JSON files in the Volume feed transactions_bronze via Auto Loader. The merchants_bronze materialized view also feeds in as the static side of the stream-static join.

In production, this lineage helps you answer questions like "if the merchant reference file had bad data, which downstream tables are affected?" The answer is immediately visible in the graph.

  • Verify you can trace the path from the raw Volume files through transactions_bronze and merchants_bronze into transactions_enriched.
  • Optionally, click on customers_history in the catalog tree and check its Lineage tab to see the CDC chain.
Query and validate business logic

The final check is to query the Silver tables directly and confirm that your joins, derived columns, deduplication, and CDC all produced correct results.

  • Open a new SQL editor tab by clicking New > Query in the workspace sidebar.
  • Count the high-value transactions by running this query:
SELECT COUNT(*) FROM fraud_detection.lakehouse.transactions_enriched WHERE is_high_value = true

What does this query confirm?

The is_high_value column is a derived field your Silver layer creates (true when amount > 5000). If this query returns a count greater than zero, the Silver enrichment logic is working correctly.

  • Verify the SCD Type 2 customer history by running this query:
SELECT * FROM fraud_detection.lakehouse.customers_history WHERE id = 'CUST001' ORDER BY __START_AT

What does this query confirm?

This query shows all historical versions of customer CUST001. Each row represents a period when that version of the customer record was active. The __START_AT and __END_AT columns define the validity window.

If a customer was deleted, the final row will have a populated __END_AT timestamp. This proves your Auto CDC with SCD Type 2 is tracking changes correctly.

  • Return to the pipeline run details and click on individual dataset nodes to review the expectations metrics.
  • Confirm that pass rates are displayed for each expectation across all tables.

Why do some expectations show less than 100% pass rate?

Expectations configured with @dp.expect (without _or_drop) log violations without removing rows. For example, valid_amount on Bronze logs any zero or negative amounts but keeps those rows for downstream investigation.

This is intentional. In production fraud detection, you want visibility into anomalous data without silently discarding it at the ingestion layer.

Query returning zero rows or an error?

If the high-value query returns zero, double-check that your Silver join completed successfully. A failed join would leave transactions_enriched empty.

If the customers_history query returns no rows for CUST001, try CUST0001 instead (the generated data uses zero-padded 4-digit IDs).

Help me troubleshoot my SQL query results.

Secret mission

Schema Evolution and Late-Arriving Data Detection

Production pipelines need to handle schema changes without breaking. They also need to detect when data arrives late. In this mission, you'll add both capabilities to your fraud detection pipeline.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs on the Databricks 14-day trial with $400 in credits. The pipeline only consumes credits when triggered, so your cost so far is minimal.

Cost warning

Your Databricks trial includes $400 in credits over 14 days. This project uses very little of that budget (a few serverless pipeline runs). If you plan to continue with additional projects, choose Keep or Pause to preserve your remaining credits for future use.

Resources you used:

  • Databricks trial workspace (14-day, $400 credits).
  • Unity Catalog: catalog fraud_detection, schema lakehouse, volume raw_data.
  • Lakeflow Declarative Pipeline: fraud_detection_pipeline.
  • Delta tables: transactions_bronze, customers_cdc_bronze, merchants_bronze, transactions_enriched, transactions_deduped, customers_cdc_clean, customers_current, customers_history.
  • Notebooks: data_generator and secret_mission_new_data.

Keep everything running

No action needed. Choose this if you plan to continue building on this pipeline or exploring Databricks features.

  • Your pipeline remains configured but only consumes credits when you manually trigger a run.
  • All tables stay queryable from the SQL editor or notebooks.
  • Monitor your remaining credits in the account console under Usage.

Pause - I'll come back to this later

Shut down the pipeline to ensure no accidental runs consume credits. Your data and tables remain intact.

  • Open your fraud_detection_pipeline in the Lakeflow Pipelines editor.
  • Confirm the pipeline is not actively running. If it is, click Stop.
  • Your tables remain queryable and your notebooks stay in the workspace. No ongoing compute cost is incurred while the pipeline is idle.

Delete - I don't want to use this again

Remove all project resources and let the trial expire naturally.

  • Delete the pipeline: open fraud_detection_pipeline in the Pipelines editor, click the three-dot menu in the top right, and select Delete.
  • Drop all Unity Catalog objects by opening a new SQL editor tab and running this command:
DROP CATALOG fraud_detection CASCADE
  • This removes the catalog, schema, volume, all Delta tables, and all data files in one command.
  • Delete the notebooks: in the workspace sidebar, right-click data_generator and select Move to Trash. Repeat for secret_mission_new_data.
  • Your Databricks trial expires after 14 days automatically. No further action is needed to close the workspace.

Nice Work!

Nice Work!

You just built a production-grade banking fraud detection data lakehouse on Databricks. From raw JSON files landing in a Unity Catalog Volume to enriched, deduplicated, and CDC-tracked tables in the Silver layer, you now have a complete medallion architecture pipeline.

Here's what you accomplished:

  • Built a Lakeflow Spark Declarative Pipeline with Bronze streaming tables ingesting banking data via Auto Loader from Unity Catalog Volumes, and Silver tables enriched with stream-static joins, window functions, and derived fraud indicators.
  • Implemented Auto CDC with both SCD Type 1 (current customer state) and SCD Type 2 (full change history with temporal tracking) for customer dimension management.
  • Enforced data quality Expectations at every layer in both logging and row-dropping modes, and traced the full data flow through Unity Catalog lineage.
  • Secret Mission: Configured Auto Loader schema evolution to handle new columns without breaking the pipeline, and added a late-arriving data expectation to detect stale records in the Silver layer.

Ready to quiz yourself?