Build a Delta Lake ETL Pipeline

Build a multi-layer ETL pipeline in Databricks using Delta Lake and PySpark.

Introduction

30 Second Summary

Every time you check your bank balance or review your transaction history, raw data from multiple systems has been cleaned, organized, and moved to where you can see it. That process of moving and transforming data is what data engineers build every day.

In this project, you will build a complete ETL pipeline in Databricks that ingests raw financial transaction data, transforms it through bronze and silver layers, and runs on a schedule.

What You'll Build

A multi-layer Delta Lake ETL pipeline in Databricks Free Edition that ingests, transforms, validates, and schedules data processing using PySpark and SQL.

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

  • A bronze Delta Lake table that stores raw transaction data exactly as it arrives, with schema evolution support for handling new fields.
  • A silver Delta Lake table with clean, validated, deduplicated records created through MERGE/upsert operations.
  • A scheduled Databricks job that orchestrates the full pipeline automatically with data quality checks.
  • Secret Mission: Build a data quality audit table that monitors data health over time.

Are there any prerequisites?

You should be comfortable with basic Python (variables, loops, functions) and basic SQL (SELECT, WHERE, JOIN). No prior Databricks or Delta Lake experience is needed.

Set Up Databricks Free Edition

Before you can build an ETL pipeline, you need a structured place to land your raw data files. Databricks organizes all data through Unity Catalog, a governance layer that gives every piece of data a clear, addressable location.

In this step, you'll explore how Unity Catalog structures your workspace and create a volume to hold the raw transaction data that feeds your pipeline.

In this step, get ready to:
  • Explore Catalog Explorer and understand the three-level namespace that organizes your data.
  • Create a managed volume named raw-data to store raw data files.
Explore Catalog Explorer

Unity Catalog uses a three-level namespace to organize data: catalog.schema.table. Your workspace comes with a workspace catalog (which shares its name with your workspace) and a default schema inside it.

  • In the Databricks sidebar, click Catalog to open Catalog Explorer.
  • In the left catalog pane, you'll see a list of catalogs available to you.
  • Click on your workspace catalog (it shares the same name as your workspace).
  • Expand it to see the default schema listed inside.

What is the three-level namespace?

Unity Catalog organizes data as catalog.schema.table. Think of a catalog as the top-level container (like a database server), a schema as a logical grouping inside it (like a database), and tables/volumes as the actual data objects inside the schema.

For Free Edition, you already have the USE CATALOG privilege on your workspace catalog and USE SCHEMA plus CREATE VOLUME privileges on the default schema by default.

You should see your workspace catalog with the default schema nested inside it. This is where all your pipeline tables and volumes will live.

Create a managed volume for raw data

A volume in Unity Catalog is a storage location for files (like CSVs, JSONs, or Parquet). You'll create a managed volume named raw-data where your pipeline's source files will live.

  • In Catalog Explorer, click Add data.
  • Select Create a volume.
  • Name the volume raw-data.
  • Select Managed volume as the volume type.
  • Select your workspace catalog and the default schema.
  • Click Create.

Why a managed volume?

A managed volume stores data in Unity Catalog's own managed storage. You don't need to configure external cloud storage or manage access credentials. Databricks handles the underlying storage for you.

This keeps your project simple and free of cloud storage costs, since Free Edition includes default managed storage.

  • Verify the volume was created by expanding your workspace catalog, then the default schema, then clicking Volumes in the schema's navigation.
  • You should see raw-data listed as a volume.

Don't see your volume?

Make sure you selected the workspace catalog (not a different catalog) and the default schema during creation. Free Edition only grants CREATE VOLUME on the default schema.

Try clicking the refresh icon in Catalog Explorer if the volume doesn't appear immediately. Help me find my volume in Catalog Explorer

Your workspace is ready with a dedicated volume for raw data. Next up, you'll create your first notebook and ingest transaction data into a bronze Delta Lake table.

Ingest Raw Data into a Bronze Delta Lake Table

Every ETL pipeline starts with ingestion. Raw data arrives from source systems and needs a landing zone where it can sit untouched before any transformations happen.

In Delta Lake architectures, this landing zone is called the bronze layer. It stores data exactly as it arrived, preserving every field and every quirk. In this step, you will create a notebook that defines raw transaction data and writes it into a bronze Delta Lake table.

In this step, get ready to:
  • Create a Python notebook for your ingestion logic.
  • Define raw transaction data inline and load it into a PySpark DataFrame.
  • Write the DataFrame to a bronze Delta Lake table and verify with SQL.
Create a new Python notebook

Each stage of your pipeline lives in its own notebook. This keeps responsibilities separated and makes the pipeline easier to schedule later.

  • In the Databricks sidebar, click + New.
  • Select Notebook from the menu.
  • A blank notebook opens with the title set to Untitled. Click the title at the top of the notebook to edit it inline.
  • Type 01_ingest_bronze and press Enter.
  • Confirm the language dropdown next to the title shows Python. On Free Edition with Unity Catalog enabled, the notebook automatically connects to serverless compute.

What is a bronze table?

In the medallion architecture (bronze, silver, gold), the bronze layer is a raw copy of source data. No filtering, no type casting, no deduplication. You preserve everything so you can always trace issues back to the source.

Later steps will clean and validate this data into a silver table.

Define raw transaction data inline

Databricks Free Edition restricts outbound internet access, so you cannot download external CSV files from code. Instead, you will define sample transaction data directly in Python.

  • Paste the following code into the first cell of your 01_ingest_bronze notebook:
# Define raw transaction data inline (simulating a CSV source)
raw_data = [
    {"transaction_id": "TXN001", "customer_id": "CUST01", "amount": 150.00, "currency": "USD", "transaction_date": "2024-01-15", "merchant": "Amazon"},
    {"transaction_id": "TXN002", "customer_id": "CUST02", "amount": 89.99, "currency": "USD", "transaction_date": "2024-01-15", "merchant": "Walmart"},
    {"transaction_id": "TXN003", "customer_id": "CUST01", "amount": -20.00, "currency": "USD", "transaction_date": "2024-01-16", "merchant": "Refund Co"},
    {"transaction_id": "TXN004", "customer_id": "CUST03", "amount": 250.50, "currency": "EUR", "transaction_date": "2024-01-16", "merchant": "Shopify"},
    {"transaction_id": "TXN005", "customer_id": None, "amount": 75.00, "currency": "USD", "transaction_date": "2024-01-17", "merchant": "Target"},
    {"transaction_id": None, "customer_id": "CUST04", "amount": 120.00, "currency": "GBP", "transaction_date": "2024-01-17", "merchant": "Tesco"},
]

# Create a PySpark DataFrame from the raw data
df = spark.createDataFrame(raw_data)

# Display the DataFrame to preview the data
display(df)

What does this code do?

  • raw_data is a list of dictionaries simulating six financial transactions. Each dictionary has six fields: transaction_id, customer_id, amount, currency, transaction_date, and merchant.
  • Some rows intentionally contain bad data (None values and a negative amount). These will be filtered out when you build the silver table in the next step.
  • spark.createDataFrame() converts the Python list into a PySpark DataFrame that Databricks can process at scale.
  • display(df) renders the data as an interactive table directly in the notebook output.
  • Run the cell by pressing Shift+Enter.
  • You should see a table with six rows and six columns rendered below the cell.

Why is the first run slow?

Databricks Free Edition uses serverless compute. The first time you run a cell, the platform provisions compute resources behind the scenes. This can take 10-30 seconds. Subsequent cells run much faster.

Cell not running?

  • Make sure the notebook language is set to Python (visible next to the notebook title). Free Edition does not support R or Scala.
  • If you see a connection error, wait a moment and try again. Serverless compute may need time to start.

Still stuck? Why is my notebook cell not executing on Databricks Free Edition?

Write the raw data to a bronze Delta Lake table

Now you will persist the DataFrame as a managed Delta Lake table. Delta Lake is the default table format in Databricks, so every table you create is automatically a Delta table.

  • Find your workspace catalog name by clicking Catalog in the sidebar. The catalog listed at the top level shares a name with your workspace.
  • Add a new cell below the first one by hovering below the cell and clicking + Code.
  • Paste the following code into the new cell, replacing the catalog name placeholder with your own workspace catalog name:
# Write raw data as-is to the bronze Delta Lake table
# Delta Lake is the default format - .format("delta") is explicit but optional
df.write.format("delta").mode("append").saveAsTable("[[CATALOG_NAME=\"your workspace catalog name\"]].default.bronze_transactions")

What does this code do?

  • .format("delta") explicitly specifies Delta Lake as the storage format. This is optional because Delta is the default, but being explicit makes the code self-documenting.
  • .mode("append") adds rows to the table without overwriting existing data. This is the standard pattern for ingestion pipelines where new batches arrive over time.
  • .saveAsTable() writes the data as a managed table using the Unity Catalog three-level namespace: catalog.schema.table.
  • Run the cell by pressing Shift+Enter.
  • The cell completes without errors. No output is displayed for write operations.

Seeing a permission error?

  • Double-check that you replaced the catalog name placeholder with your actual workspace catalog name (visible in the Catalog sidebar).
  • Make sure you are writing to the default schema. Free Edition users have CREATE TABLE privileges on the workspace catalog and default schema by default.

Need more help? Why am I getting a permission error when writing to my Delta table?

Query the bronze table to verify the data

Time to confirm the data landed correctly. You can run SQL directly in a Python notebook using the %sql magic command at the start of a cell.

  • Add a new cell below and paste the following SQL query:
%sql
-- Verify the bronze table contains the ingested rows
SELECT * FROM default.bronze_transactions

What does %sql do?

The %sql magic command switches a single cell's language to SQL, even though the notebook default is Python. This lets you query tables using familiar SQL syntax without leaving the notebook.

The query uses the two-level namespace default.bronze_transactions (schema.table). This works because the workspace catalog is already set as your default catalog.

  • Run the cell by pressing Shift+Enter.
  • You should see all six rows displayed in a table with columns: transaction_id, customer_id, amount, currency, transaction_date, and merchant.

Table not found?

  • Make sure the previous write cell completed successfully (no red error output).
  • Verify the cell starts with %sql on the very first line. The magic command must be the first content in the cell with nothing above it.

Still stuck? Help me troubleshoot a table not found error in Databricks.

✔️ Awesome, I've got everything!

Great. Databricks auto-saves your notebook, but you can confirm by checking the last-edit timestamp near the title.

ⓧ I'd like to double check the full code

Here is the complete 01_ingest_bronze notebook. Each code block below represents a separate cell:

# Define raw transaction data inline (simulating a CSV source)
raw_data = [
    {"transaction_id": "TXN001", "customer_id": "CUST01", "amount": 150.00, "currency": "USD", "transaction_date": "2024-01-15", "merchant": "Amazon"},
    {"transaction_id": "TXN002", "customer_id": "CUST02", "amount": 89.99, "currency": "USD", "transaction_date": "2024-01-15", "merchant": "Walmart"},
    {"transaction_id": "TXN003", "customer_id": "CUST01", "amount": -20.00, "currency": "USD", "transaction_date": "2024-01-16", "merchant": "Refund Co"},
    {"transaction_id": "TXN004", "customer_id": "CUST03", "amount": 250.50, "currency": "EUR", "transaction_date": "2024-01-16", "merchant": "Shopify"},
    {"transaction_id": "TXN005", "customer_id": None, "amount": 75.00, "currency": "USD", "transaction_date": "2024-01-17", "merchant": "Target"},
    {"transaction_id": None, "customer_id": "CUST04", "amount": 120.00, "currency": "GBP", "transaction_date": "2024-01-17", "merchant": "Tesco"},
]

# Create a PySpark DataFrame from the raw data
df = spark.createDataFrame(raw_data)

# Display the DataFrame to preview the data
display(df)
# Write raw data as-is to the bronze Delta Lake table
# Delta Lake is the default format - .format("delta") is explicit but optional
df.write.format("delta").mode("append").saveAsTable("[[CATALOG_NAME=\"your workspace catalog name\"]].default.bronze_transactions")
%sql
-- Verify the bronze table contains the ingested rows
SELECT * FROM default.bronze_transactions

Your bronze layer is live. Raw transaction data is now stored in a managed Delta Lake table, exactly as it arrived. Next up, you will build a silver table that cleans, validates, and deduplicates this data using transformations and MERGE operations.

Transform and Upsert into a Silver Table

Your bronze table now holds raw transaction data exactly as it arrived. Raw data is messy though. Amounts might be stored as strings, dates might be inconsistently formatted, and invalid records could be lurking in the rows.

In this step, you will build the silver layer of your pipeline. The silver layer cleans, validates, and deduplicates data so downstream consumers can trust it. You will use PySpark transformations to fix data types, filter out bad records, and then use a MERGE operation to upsert records into the silver table.

In this step, get ready to:
  • Read bronze data and apply type casting, filtering, and timestamp enrichment.
  • Upsert cleaned records into a silver Delta Lake table using the DeltaTable MERGE API.
  • Verify the silver table contains only valid, deduplicated records with correct data types.
Create the 02_transform_silver notebook

Just like the ingestion notebook, you will create a dedicated notebook for the silver transformation logic. Keeping each pipeline stage in its own notebook makes it easy to orchestrate them as separate tasks later.

  • In your Databricks workspace sidebar, click + New.
  • Select Notebook from the menu.
  • A blank notebook opens. Click the notebook title at the top (it will say something like Untitled Notebook) and rename it to 02_transform_silver.
  • Confirm the language is set to Python (the default if your last notebook was Python).

What about compute?

Since your workspace has Unity Catalog enabled, the notebook automatically connects to serverless compute. You do not need to create or attach a cluster.

Read from bronze and apply transformations

The bronze table stores raw strings and unvalidated values. The silver layer needs typed, validated data. You will cast amount to a decimal, parse transaction_date into a proper date, filter out invalid records, and stamp each row with a processing timestamp.

  • In the first cell of your 02_transform_silver notebook, paste the following code:
from pyspark.sql.functions import col, current_timestamp, to_date
from pyspark.sql.types import DecimalType

# Read raw data from the bronze table
bronze_df = spark.read.table("workspace_catalog.default.bronze_transactions")

# Apply transformations: cast types, add processing timestamp
silver_df = (
    bronze_df
    .withColumn("amount", col("amount").cast(DecimalType(10, 2)))
    .withColumn("transaction_date", to_date(col("transaction_date"), "yyyy-MM-dd"))
    .withColumn("processed_at", current_timestamp())
)

# Filter out invalid records: null transaction_id or negative amounts
silver_df = silver_df.filter(
    (col("transaction_id").isNotNull()) & (col("amount") >= 0)
)

silver_df.display()
  • Press Shift+Enter to run the cell.

What does this code do?

  • spark.read.table() reads the bronze Delta Lake table into a PySpark DataFrame.
  • .cast(DecimalType(10, 2)) converts the amount from a string into a decimal number with 2 decimal places.
  • to_date() parses the transaction_date string into a proper date type so you can sort and filter by date.
  • current_timestamp() stamps each row with the exact time it was processed, creating an audit trail.
  • The .filter() call removes any rows with a null transaction_id or a negative amount. These are invalid records that should not reach the silver layer.

You should see a table output showing your transformed rows with the amount column as a decimal, transaction_date as a date, and a new processed_at column with today's timestamp.

Seeing an error about the table not found?

Make sure you ran all the cells in your 01_ingest_bronze notebook first. The bronze table must exist before the silver notebook can read from it.

Double check the table name matches exactly: workspace_catalog.default.bronze_transactions. Your workspace catalog name should match the name shown in Catalog Explorer.

Help me fix a table not found error in my Databricks silver transformation notebook.

Upsert into the silver table with MERGE

A simple write would create duplicates every time the pipeline runs. Instead, you will use a MERGE operation (also called an upsert). MERGE matches incoming rows against existing rows by a key. If a match is found, it updates the existing record. If no match is found, it inserts a new one.

  • Click the + Code button below your first cell to add a new cell, then paste the following code:
from delta.tables import DeltaTable

# Table name for the silver layer
silver_table_name = "workspace_catalog.default.silver_transactions"

# Check if the silver table already exists
if not spark.catalog.tableExists(silver_table_name):
    # First run: create the silver table from the transformed DataFrame
    silver_df.write.format("delta").saveAsTable(silver_table_name)
else:
    # Subsequent runs: upsert using MERGE on transaction_id
    delta_table = DeltaTable.forName(spark, silver_table_name)

    (delta_table.alias("target")
        .merge(silver_df.alias("source"), "target.transaction_id = source.transaction_id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute()
    )

print("Silver table upsert complete.")
  • Press Shift+Enter to run the cell.

What does this code do?

  • from delta.tables import DeltaTable imports the Delta Lake Python API that enables MERGE operations.
  • spark.catalog.tableExists() checks whether the silver table has been created yet. On the first run, the table does not exist, so the code creates it.
  • DeltaTable.forName() gets a reference to the existing silver table for the MERGE operation.
  • The .merge() call matches source rows to target rows on transaction_id. .whenMatchedUpdateAll() updates existing rows, and .whenNotMatchedInsertAll() inserts new ones.

Why MERGE instead of overwrite?

Overwriting the entire table is simple, but it destroys history and forces a full rewrite. MERGE only touches the rows that changed. In production pipelines processing millions of rows, this saves significant time and compute.

MERGE also handles idempotency. If the pipeline runs twice on the same data, existing rows get updated in place rather than duplicated.

You should see the message Silver table upsert complete. printed below the cell.

Seeing a permission or table creation error?

Confirm you are using the workspace catalog name (it shares the name of your workspace) in the table path. You can check this by clicking Catalog in the sidebar and looking at the top-level catalog listed.

Help me debug a MERGE or table creation error in my Databricks silver notebook.

Verify the silver table

The final check queries the silver table directly to confirm the transformations landed correctly and no duplicates snuck through.

  • Add a new cell and paste the following SQL verification query:
# Verify the silver table: check schema and row contents
result = spark.sql("SELECT * FROM workspace_catalog.default.silver_transactions")
result.display()

# Print the schema to confirm data types
result.printSchema()
  • Press Shift+Enter to run the cell.

What should I see?

You should see a table with your transaction rows. The amount column displays as a decimal (e.g. 49.99), transaction_date shows as a date (e.g. 2024-01-15), and the processed_at column has a full timestamp. Any rows with null IDs or negative amounts from your bronze data are gone.

The schema printout confirms the data types: amount as decimal(10,2), transaction_date as date, and processed_at as timestamp.

✔️ Awesome, I've got everything!

Great. Double check you have all three cells in your notebook, then move on to the next step.

ⓧ I'd like to double check the full code

Here is the complete 02_transform_silver notebook with all three cells:

# Cell 1: Read from bronze and apply transformations
from pyspark.sql.functions import col, current_timestamp, to_date
from pyspark.sql.types import DecimalType

# Read raw data from the bronze table
bronze_df = spark.read.table("workspace_catalog.default.bronze_transactions")

# Apply transformations: cast types, add processing timestamp
silver_df = (
    bronze_df
    .withColumn("amount", col("amount").cast(DecimalType(10, 2)))
    .withColumn("transaction_date", to_date(col("transaction_date"), "yyyy-MM-dd"))
    .withColumn("processed_at", current_timestamp())
)

# Filter out invalid records: null transaction_id or negative amounts
silver_df = silver_df.filter(
    (col("transaction_id").isNotNull()) & (col("amount") >= 0)
)

silver_df.display()
# Cell 2: Upsert into silver table using MERGE
from delta.tables import DeltaTable

# Table name for the silver layer
silver_table_name = "workspace_catalog.default.silver_transactions"

# Check if the silver table already exists
if not spark.catalog.tableExists(silver_table_name):
    # First run: create the silver table from the transformed DataFrame
    silver_df.write.format("delta").saveAsTable(silver_table_name)
else:
    # Subsequent runs: upsert using MERGE on transaction_id
    delta_table = DeltaTable.forName(spark, silver_table_name)

    (delta_table.alias("target")
        .merge(silver_df.alias("source"), "target.transaction_id = source.transaction_id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute()
    )

print("Silver table upsert complete.")
# Cell 3: Verify the silver table
result = spark.sql("SELECT * FROM workspace_catalog.default.silver_transactions")
result.display()

# Print the schema to confirm data types
result.printSchema()

Your silver table is live with clean, deduplicated data. Next up, you will handle schema evolution when new columns arrive in your source data.

Implement Schema Evolution

Your bronze and silver tables are working perfectly for the current data shape. But in production, data sources change all the time.

New fields get added, columns get renamed, and your pipeline needs to handle these changes without breaking. In this step, you will simulate a schema change by adding a new category field to incoming data and teach both layers of your pipeline to evolve gracefully.

In this step, get ready to:
  • Write a second batch of data with a new column to the bronze table using schema evolution.
  • Update the silver transformation to handle new columns with MERGE WITH SCHEMA EVOLUTION.
  • Verify that old and new records coexist with the evolved schema.
Write a second batch with a new field to the bronze table

Imagine the upstream system starts including a category field with each transaction. Your bronze table does not have this column yet.

  • Switch back to your 01_ingest_bronze notebook in your Databricks workspace.
  • Hover below the last cell and click + Code to add a new code cell.
  • Define and write a second batch of transactions (including the new category column) by pasting this code into the new cell:
# Define a second batch of transactions with a new 'category' field
new_data = [
    ("TXN006", "C003", 250.00, "USD", "2024-01-06", "TechStore", "electronics"),
    ("TXN007", "C001", 45.00, "USD", "2024-01-07", "GroceryMart", "groceries"),
    ("TXN008", "C004", 180.00, "EUR", "2024-01-08", "FashionHub", "clothing"),
]

# Column names now include the new 'category' field
new_columns = ["transaction_id", "customer_id", "amount", "currency", "transaction_date", "merchant", "category"]

# Create a DataFrame from the new batch
new_df = spark.createDataFrame(new_data, new_columns)

# Write to the bronze table with schema evolution enabled
new_df.write \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("workspace_catalog.default.bronze_transactions")

print("Second batch written to bronze with schema evolution!")

What does this code do?

The new batch includes a seventh field, category, that the original batch did not have. Without special handling, writing this to the existing table would fail because the schemas do not match.

The .option("mergeSchema", "true") setting tells Delta Lake to automatically add the new column to the table schema during the write. Existing rows get a null value for category because they were written before the column existed.

  • Run the cell by pressing Shift+Enter.
  • You should see the output: Second batch written to bronze with schema evolution!.

Seeing a schema mismatch error?

Make sure you typed .option("mergeSchema", "true") exactly as shown. The option name is case-sensitive. Also confirm you are using mode("append") and not mode("overwrite").

If you see a different error, help me debug this schema evolution error in my Databricks notebook.

  • Add another code cell below and verify the bronze table now includes the category column by running this query:
# Verify the bronze table schema has evolved to include 'category'
result = spark.sql("SELECT * FROM workspace_catalog.default.bronze_transactions")
result.display()

What should I see?

The results table now has seven columns. The original rows (TXN001 through TXN005) show null in the category column. The new rows (TXN006 through TXN008) show values like electronics, groceries, and clothing.

Update the silver transformation to handle the new column

The bronze table now has a category column, but your silver table does not. You need to update the MERGE operation so it evolves the silver schema automatically when new columns appear.

  • Switch back to your 02_transform_silver notebook.
  • Find the cell that contains your existing transformation and MERGE logic.
  • Replace the entire contents of that cell with this updated version that uses .withSchemaEvolution() on the MERGE operation:
from delta.tables import DeltaTable
from pyspark.sql.functions import current_timestamp, col
from pyspark.sql.types import DecimalType, DateType

# Read all records from the bronze table
bronze_df = spark.read.table("workspace_catalog.default.bronze_transactions")

# Apply transformations: cast types, filter invalid records, add processed_at
silver_df = (
    bronze_df
    .withColumn("amount", col("amount").cast(DecimalType(10, 2)))
    .withColumn("transaction_date", col("transaction_date").cast(DateType()))
    .filter(col("transaction_id").isNotNull())
    .filter(col("amount") >= 0)
    .withColumn("processed_at", current_timestamp())
)

# Upsert into silver table with schema evolution enabled
silver_table = DeltaTable.forName(spark, "workspace_catalog.default.silver_transactions")

(
    silver_table.alias("target")
    .merge(silver_df.alias("source"), "target.transaction_id = source.transaction_id")
    .withSchemaEvolution()
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute()
)

print("Silver table updated with schema evolution!")

What does this code do?

The transformation logic is the same as before: cast types, filter invalid records, and add the processed_at timestamp. The key difference is the .withSchemaEvolution() method added to the MERGE chain.

This tells Delta Lake to automatically add any new columns from the source DataFrame to the target table during the merge. The category column gets added to the silver table without you having to manually alter the schema.

  • Run the cell by pressing Shift+Enter.
  • You should see: Silver table updated with schema evolution!.

Getting an AnalysisException?

Ensure you are calling .withSchemaEvolution() before .whenMatchedUpdateAll() in the method chain. The order matters. Also check that your silver table exists from the previous step.

If the error persists, help me fix this MERGE WITH SCHEMA EVOLUTION error.

Verify the evolved silver table schema

Now confirm that both the schema evolved and the data is correct. Old records should have null for category while new records have it populated.

  • Add a new code cell below in the 02_transform_silver notebook and run this schema check:
# Check the silver table schema to confirm the new 'category' column exists
spark.sql("DESCRIBE workspace_catalog.default.silver_transactions").display()

What should I see?

The DESCRIBE output shows all columns in your silver table. You should see category listed as a column with type string, alongside the existing columns: transaction_id, customer_id, amount, currency, transaction_date, merchant, and processed_at.

  • Add one more code cell and run this query to verify the data:
# Verify old records have null category, new records have category populated
spark.sql("""
    SELECT transaction_id, customer_id, merchant, category
    FROM workspace_catalog.default.silver_transactions
    ORDER BY transaction_id
""").display()

What should I see?

The query results show all your transactions ordered by ID. Records TXN001 through TXN005 have null in the category column because they were written before the column existed. Records TXN006 through TXN008 show their category values: electronics, groceries, and clothing.

Why is schema evolution important?

In production, upstream data sources add new fields regularly. Without schema evolution, your pipeline would break every time the source schema changes. You would have to manually alter the table and redeploy.

With .option("mergeSchema", "true") for writes and .withSchemaEvolution() for merges, your pipeline adapts automatically. This is a core requirement for any enterprise ETL system.

✔️ Awesome, I've got everything!

Great. Double check that both notebooks are saved and your silver table shows the category column in the DESCRIBE output.

ⓧ I'd like to double check the full code

Here is the new cell you added to 01_ingest_bronze:

# Define a second batch of transactions with a new 'category' field
new_data = [
    ("TXN006", "C003", 250.00, "USD", "2024-01-06", "TechStore", "electronics"),
    ("TXN007", "C001", 45.00, "USD", "2024-01-07", "GroceryMart", "groceries"),
    ("TXN008", "C004", 180.00, "EUR", "2024-01-08", "FashionHub", "clothing"),
]

# Column names now include the new 'category' field
new_columns = ["transaction_id", "customer_id", "amount", "currency", "transaction_date", "merchant", "category"]

# Create a DataFrame from the new batch
new_df = spark.createDataFrame(new_data, new_columns)

# Write to the bronze table with schema evolution enabled
new_df.write \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("workspace_catalog.default.bronze_transactions")

print("Second batch written to bronze with schema evolution!")

Here is the updated transformation cell in 02_transform_silver:

from delta.tables import DeltaTable
from pyspark.sql.functions import current_timestamp, col
from pyspark.sql.types import DecimalType, DateType

# Read all records from the bronze table
bronze_df = spark.read.table("workspace_catalog.default.bronze_transactions")

# Apply transformations: cast types, filter invalid records, add processed_at
silver_df = (
    bronze_df
    .withColumn("amount", col("amount").cast(DecimalType(10, 2)))
    .withColumn("transaction_date", col("transaction_date").cast(DateType()))
    .filter(col("transaction_id").isNotNull())
    .filter(col("amount") >= 0)
    .withColumn("processed_at", current_timestamp())
)

# Upsert into silver table with schema evolution enabled
silver_table = DeltaTable.forName(spark, "workspace_catalog.default.silver_transactions")

(
    silver_table.alias("target")
    .merge(silver_df.alias("source"), "target.transaction_id = source.transaction_id")
    .withSchemaEvolution()
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute()
)

print("Silver table updated with schema evolution!")

Here are the verification cells in 02_transform_silver:

# Check the silver table schema to confirm the new 'category' column exists
spark.sql("DESCRIBE workspace_catalog.default.silver_transactions").display()
# Verify old records have null category, new records have category populated
spark.sql("""
    SELECT transaction_id, customer_id, merchant, category
    FROM workspace_catalog.default.silver_transactions
    ORDER BY transaction_id
""").display()

Your pipeline now adapts to schema changes automatically. Next up, you will add data quality checks and schedule the entire pipeline to run as a Databricks job.

Add Data Quality Checks and Schedule the Pipeline

Your silver table now contains clean, deduplicated transaction records with an evolved schema. But in a production pipeline, data quality can degrade silently over time.

In this step, you will add automated quality checks that halt the pipeline if bad data sneaks through. Then you will wire everything together as a scheduled Databricks job that runs daily without manual intervention.

In this step, get ready to:
  • Add data quality assertions to the silver transformation notebook.
  • Create a multi-task Databricks job with task dependencies.
  • Schedule the job and verify a successful manual run.
Add data quality assertions to the silver notebook

Quality checks act as guardrails that stop bad data from propagating downstream. If a check fails, the notebook raises an error and the job stops before damaged data reaches consumers.

  • Switch back to your 02_transform_silver notebook in the Databricks workspace.
  • Add a new cell at the bottom of the notebook with the following data quality checks:
# -----------------------------------------------
# Data quality checks on the silver table
# -----------------------------------------------
silver_df = spark.read.table("workspace_catalog.default.silver_transactions")

# Check 1: No null transaction_ids (primary key integrity)
null_count = silver_df.filter("transaction_id IS NULL").count()
assert null_count == 0, f"Quality check failed: {null_count} null transaction_ids found"

# Check 2: All amounts within valid range (0 to 1,000,000)
invalid_amounts = silver_df.filter("amount < 0 OR amount > 1000000").count()
assert invalid_amounts == 0, f"Quality check failed: {invalid_amounts} amounts out of valid range"

# Check 3: No duplicate transaction_ids (uniqueness check)
total_count = silver_df.count()
distinct_count = silver_df.select("transaction_id").distinct().count()
assert total_count == distinct_count, f"Quality check failed: {total_count - distinct_count} duplicate transaction_ids found"

print("All quality checks passed!")

What does this code do?

Each check reads the silver table and uses a Python assert statement to halt execution if the condition is violated.

  • Check 1 ensures no rows have a null transaction_id. A null primary key would break downstream joins and reporting.
  • Check 2 validates that all amounts fall within a reasonable range. Negative amounts were already filtered, but this catches any edge cases that slipped through.
  • Check 3 compares total row count against distinct transaction_id count. If they differ, duplicates exist despite the MERGE upsert logic.
  • Run the new cell by clicking the play button or pressing Shift+Enter.

You should see the output All quality checks passed! printed below the cell.

Seeing an assertion error?

If a check fails, it means your silver table has unexpected data. Re-run the 02_transform_silver notebook from the top to re-apply transformations and the MERGE logic. Then re-run the quality checks cell.

Help me debug a failed data quality assertion in my silver table.

Create a scheduled Databricks job

A Databricks job orchestrates multiple notebooks as tasks with defined dependencies. This means the silver transformation only runs after ingestion succeeds.

  • Click Jobs and Pipelines in the left sidebar.
  • Click Create, then click Job.
  • Click the Notebook tile to configure the first task.
  • Replace the default job name at the top with ETL Pipeline.
  • In the Task name field, enter ingest_bronze.
  • In the Source dropdown, select Workspace.
  • For Path, use the file browser to find and select your 01_ingest_bronze notebook, then click Confirm.
  • Click Create task.

What about compute?

Your workspace uses serverless compute by default because Unity Catalog is enabled. You do not need cluster creation permission.

Databricks provisions and manages the compute automatically for each task.

Now add the second task for the silver transformation.

  • Click + Add task and select Notebook.
  • In the Task name field, enter transform_silver.
  • For Path, browse to and select your 02_transform_silver notebook, then click Confirm.
  • Verify the Depends on field shows ingest_bronze. This sets up the dependency so the transform runs only after ingestion succeeds.
  • Click Create task.

Why task dependencies?

Dependencies create a Directed Acyclic Graph (DAG) that controls execution order. The silver transformation reads from the bronze table, so it must wait until ingestion finishes writing new data.

If ingestion fails, the transform task is skipped automatically.

Now add a daily schedule so the pipeline runs automatically.

  • In the Job details pane on the right, scroll down to the Schedules & Triggers section.
  • Click Add trigger.
  • Under Trigger type, choose Scheduled.
  • In Schedule type, select Simple.
  • Set the interval to 1 and the unit to Days.
  • Click Save.

Pausing the schedule

You can pause the schedule at any time from the Schedules & Triggers section by clicking Pause. This stops automatic runs without deleting the job.

Trigger a manual run and verify success

Before relying on the daily schedule, run the job manually to confirm everything works end to end.

  • Click Run now in the upper right corner of the job page.
  • Wait for the run to complete. This may take a minute or two while serverless compute spins up.
  • Click the Runs tab.
  • Click the link in the Start time column to open the run details.

You should see both tasks with green checkmarks indicating successful completion. The transform_silver task ran after ingest_bronze completed, and your quality checks passed without assertion errors.

Task showing a red X or failed status?

Click the failed task name to see the error output. Common causes:

  • A quality assertion failed because duplicate data was appended. Drop the bronze table with DROP TABLE default.bronze_transactions in a SQL cell, re-run 01_ingest_bronze once, then re-run the job.
  • A notebook path was not found. Edit the task and re-select the notebook using the file browser.

Help me debug a failed task in my Databricks ETL job.

Your ETL pipeline is now fully automated. Next up, you will take on the Secret Mission and build a data quality audit table that tracks pipeline health over time.

Secret mission

Build a Data Quality Audit Table

Your pipeline catches bad data and raises errors. But in a real data team, you need more than pass/fail. You need a historical record of every quality check, so you can spot trends, catch gradual degradation, and prove compliance. Build a monitoring notebook that writes audit results to a dedicated Delta table.

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 Databricks Free Edition, so there are no ongoing costs. Your tables and job persist as long as you use the account periodically.

Resources you used:

  • Delta Lake tables: bronze_transactions, silver_transactions, quality_audit (stored in managed Unity Catalog storage).
  • Managed volume: raw-data in the workspace catalog, default schema.
  • Scheduled Databricks job (ETL pipeline with three tasks).
  • Three notebooks: 01_ingest_bronze, 02_transform_silver, 03_quality_audit.
  • Databricks Free Edition account (free, no expiry within fair usage quotas).

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the pipeline soon.

  • Your Free Edition account persists indefinitely with no charges.
  • Tables and the scheduled job remain as long as you use the workspace periodically.
  • The daily job schedule will continue to trigger runs, consuming your daily serverless quota. If you want to conserve quota, consider pausing the schedule (see the Pause tab).

Pause - I'll come back to this later

Pause the scheduled job trigger to stop automatic daily runs. Your tables and notebooks stay intact for when you return.

  • Click Jobs & Pipelines in the sidebar.
  • Click the name of your ETL pipeline job to open its job page.
  • In the Schedules & Triggers section, click the toggle to pause the schedule.
  • Your tables, volume, and notebooks remain untouched. Re-enable the schedule whenever you are ready to resume.

Delete - I don't want to use this again

Remove all project resources and start fresh. You can keep the Databricks account itself for future learning.

Drop the Delta Lake tables by running the following SQL statements in any notebook:

DROP TABLE IF EXISTS default.bronze_transactions;
DROP TABLE IF EXISTS default.silver_transactions;
DROP TABLE IF EXISTS default.quality_audit;

Drop the managed volume:

DROP VOLUME IF EXISTS default.raw_data;
  • Delete the scheduled job: click Jobs & Pipelines in the sidebar, click the job name to open its page, click the kebab menu icon next to the job name, and select Delete job.
  • Delete each notebook (01_ingest_bronze, 02_transform_silver, 03_quality_audit): open the notebook, click the kebab menu at the top right, and select Move to Trash.
  • Your Databricks Free Edition account can remain active at no cost for future projects.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-style Delta Lake ETL pipeline in Databricks that ingests, transforms, validates, and schedules data processing automatically.

You've learned how to:

  • Built a multi-layer Delta Lake pipeline (bronze and silver tables) using PySpark in Databricks notebooks, with MERGE/upsert operations for deduplication.
  • Implemented schema evolution to handle new columns arriving in source data without breaking your pipeline or losing existing records.
  • Scheduled the entire pipeline as a multi-task Databricks job with data quality assertions that catch issues before they propagate downstream.
  • Secret Mission: Built a quality audit table that tracks data health over time, giving you a monitoring pattern used in production data platforms.

Ready to quiz yourself?