Build a Delta Lake Lakehouse
Build a medallion lakehouse with Delta Lake on Databricks Free Edition.
Introduction
30 Second Summary
Every time you buy something online, a chain of data events fires off behind the scenes. Orders, payments, reviews, and shipping updates all need to land in a system that keeps them accurate, consistent, and ready to query. When that system breaks mid-write, you get corrupted data and broken dashboards.
In this project, you will build a complete data lakehouse using Delta Lake on Databricks Free Edition. You'll process real Brazilian e-commerce data through Bronze, Silver, and Gold layers while mastering features like time travel, schema evolution, and MERGE operations.
What You'll Build
A Medallion Architecture lakehouse processing over 100,000 Brazilian e-commerce records through Delta Lake on Databricks Free Edition.
By the end of this project, you'll have:
- A working Medallion Architecture pipeline (Bronze, Silver, Gold) processing real e-commerce data with ACID guarantees.
- Hands-on demonstrations of Delta Lake features including time travel, schema evolution, and MERGE/upsert operations.
- A GitHub-ready repository with organized notebooks exported as Python files and a comprehensive README.
- Secret Mission: Build a Change Data Feed pipeline that tracks row-level changes across your Silver layer and creates an audit log.
Are there any prerequisites?
This project assumes you're comfortable writing Python and SQL. You should also have a basic understanding of Medallion Architecture concepts (Bronze, Silver, and Gold layers).
Set Up Databricks and Download the Dataset
Before you can build a lakehouse, you need a platform to run your data pipelines on and data to process.
In this step, you'll create a free Databricks workspace and download a real-world e-commerce dataset from Kaggle. You'll then upload the raw CSV files into a Unity Catalog volume so they're accessible from your notebooks.
In this step, get ready to:
- Sign up for Databricks Free Edition and verify your workspace.
- Download the Brazilian E-Commerce dataset from Kaggle.
- Create a Unity Catalog volume and upload all CSV files.
Sign up for Databricks Free Edition
Databricks Free Edition is a no-cost, serverless workspace designed for learning and experimentation. It replaced the legacy Community Edition, which was retired in 2025.
- Open the Databricks signup page in your browser:
- https://www.databricks.com/try-databricks
- Under the FOR PERSONAL USE section, click Get Free Edition.
- Choose your preferred signup method (email or a social login).
- Complete the registration form and submit it.
Databricks will create a new workspace for you automatically. This may take a minute.
- Once your workspace loads, confirm you can see the sidebar on the left with Catalog, Compute, and Notebooks sections.
What is Databricks Free Edition?
Databricks Free Edition gives you a serverless workspace with daily usage quotas. It supports Python and SQL only. No cloud account or credit card is required.
If you exceed your daily quota, compute resources shut down until the next day. Your data and settings are never deleted.
Previously used Community Edition?
Community Edition was retired in 2025. If you still have a Community Edition workspace, use the migration banner at the top of your CE workspace to move to Free Edition. Otherwise, sign up for a fresh Free Edition account.
help me migrate from Community Edition to Free Edition
Download the Brazilian E-Commerce dataset from Kaggle
The dataset you'll use is the Brazilian E-Commerce Public Dataset by Olist. It contains real transactional data from a Brazilian marketplace with approximately 100,000 orders across 8 interconnected CSV files.
- Open the dataset page on Kaggle:
- kaggle.com/datasets/olistbr/brazilian-ecommerce
- Click Download to download the full dataset as a zip file. You'll need a free Kaggle account to download.
- Extract the zip file on your computer.
After extracting, confirm you have these 8 CSV files:
- olist_orders_dataset.csv
- olist_order_items_dataset.csv
- olist_customers_dataset.csv
- olist_products_dataset.csv
- olist_order_payments_dataset.csv
- olist_order_reviews_dataset.csv
- olist_sellers_dataset.csv
- product_category_name_translation.csv
Why this dataset?
The Olist dataset has multiple related tables (orders, customers, products, payments, reviews) that mirror real production data. This makes it perfect for demonstrating MERGE operations, multi-table joins, and schema evolution across a Medallion Architecture.
Can't download from Kaggle?
Kaggle requires a free account to download datasets. Sign up at kaggle.com if you don't already have one. Make sure you're signed in before clicking Download.
help me download a dataset from Kaggle
Create a Unity Catalog volume and upload your data
Databricks Free Edition does not support DBFS mounts. Instead, you'll use a Unity Catalog managed volume to store your CSV files. This is the current production pattern for data storage on Databricks.
- In your Databricks workspace, click Catalog in the left sidebar to open Catalog Explorer.
- Click Add data.
- Select Create a volume.
- Name the volume olist-data.
- Select Managed as the volume type.
- Choose the workspace catalog and default schema.
- Click Create.
Now upload your CSV files to the volume.
- Open the newly created olist-data volume in Catalog Explorer.
- Click Upload to this volume.
- Drag and drop all 8 CSV files from your extracted folder, or click to browse and select them.
- Wait for all uploads to complete.
- Verify that all 8 files appear in the volume. You should see each CSV listed under the path /Volumes/workspace/default/olist-data/.
Why a Unity Catalog volume instead of DBFS?
Databricks Free Edition does not support DBFS mounts. Unity Catalog volumes are the recommended approach for storing files in Databricks. They provide governed access to data files and work seamlessly with the serverless compute environment.
Your files are now accessible from any notebook at the path /Volumes/workspace/default/olist-data/<filename>.
Upload failing or taking too long?
Make sure you're uploading the individual CSV files, not the zip archive itself. Each file should be under 10 MB. If uploads stall, try uploading 2-3 files at a time rather than all 8 at once.
help me upload files to a Unity Catalog volume
Your workspace is ready and your data is in place. Next up, you'll create your first notebook and explore how Delta Lake compares to plain Parquet.
Understand Delta Lake vs Parquet
Your Databricks workspace is set up and your dataset is ready. Now it is time to understand the fundamental difference between storing data as plain Parquet files and storing data as Delta Lake tables.
On the surface, Delta Lake looks like Parquet. Under the hood, it adds a transaction log that unlocks features Parquet alone cannot provide. In this step, you will write the same data in both formats and prove the difference through hands-on experiments.
In this step, get ready to:
- Write orders data as both a Parquet table and a Delta table, then inspect their file structures.
- Demonstrate that UPDATE fails on Parquet but succeeds on Delta.
- Demonstrate that time travel works on Delta but is impossible on Parquet.
Create the comparison notebook and write both formats
You need to see the raw file structure of both formats side by side. To make this possible, you will write the data to your Unity Catalog volume as files, so you can use dbutils.fs.ls() to list what is inside each directory.
- In the Databricks sidebar, click New and select Notebook.
- Name it 01_delta_vs_parquet and make sure the default language is set to Python.
- In the first cell, load the orders CSV from your volume and preview it by running this code:
# Load orders data from the Unity Catalog volume
orders_df = spark.read.csv(
"/Volumes/workspace/default/olist-data/olist_orders_dataset.csv",
header=True,
inferSchema=True
)
# Preview the first few rows
display(orders_df.limit(5))What does this code do?
spark.read.csv() reads the CSV file from your olist-data volume. The header=True option tells Spark the first row contains column names. inferSchema=True lets Spark detect data types automatically.
display() renders the DataFrame as a table in your notebook output.
- Run the cell by pressing Shift+Enter. You should see a table with columns like order_id, customer_id, order_status, and several timestamp columns.
Not seeing any data?
Confirm the file path matches exactly: /Volumes/workspace/default/olist-data/olist_orders_dataset.csv. The volume name must match what you created in Step 1. Also check that your notebook is connected to compute. The first cell may take a moment while serverless compute starts up.
Help me debug why my notebook can't read the CSV from my volume.
- In a new cell, write the orders data as Parquet files and as Delta files to your volume. Run this code:
# Write orders as plain Parquet files to the volume
parquet_path = "/Volumes/workspace/default/olist-data/orders_parquet"
orders_df.write.format("parquet").mode("overwrite").save(parquet_path)
# Write orders as Delta files to the volume
delta_path = "/Volumes/workspace/default/olist-data/orders_delta"
orders_df.write.format("delta").mode("overwrite").save(delta_path)
print("Both formats written successfully!")What does this code do?
Both writes save the same data to subdirectories in your volume. The only difference is the format parameter: one is parquet and the other is delta. This lets you compare the raw file structures side by side.
- Run the cell. You should see the message Both formats written successfully! in the output.
- Now inspect both directories to see the structural difference. Add a new cell and run:
# List files in the Parquet directory
print("=== Parquet directory ===")
display(dbutils.fs.ls("/Volumes/workspace/default/olist-data/orders_parquet"))
print("\n=== Delta directory ===")
display(dbutils.fs.ls("/Volumes/workspace/default/olist-data/orders_delta"))What should I see?
The Parquet directory shows only .parquet data files and a _SUCCESS marker. The Delta directory shows the same .parquet data files PLUS a _delta_log/ directory. This extra directory is the transaction log. It is the single thing that separates Delta Lake from plain Parquet.
- Inspect the _delta_log directory itself. Add a new cell and run:
# Look inside the _delta_log directory
display(dbutils.fs.ls("/Volumes/workspace/default/olist-data/orders_delta/_delta_log"))What is the _delta_log?
You will see JSON files like 00000000000000000000.json. Each JSON file represents one version of the table. Every write, update, or delete creates a new entry in this log.
This transaction log is what enables ACID transactions, time travel, schema evolution, and DML operations (UPDATE, DELETE, MERGE). Without it, Parquet is just a static file format.
- Run the cell. You should see at least one JSON file (version 0) listed inside the _delta_log directory.
Seeing a FileNotFoundException?
Double check the path is typed exactly as /Volumes/workspace/default/olist-data/orders_delta/_delta_log. If the Delta write cell above failed silently, re-run it first. You must run cells in order from top to bottom.
Help me figure out why I can't see the _delta_log directory.
Demonstrate that UPDATE fails on Parquet but succeeds on Delta
The transaction log is not just metadata. It is what allows Delta Lake to support row-level changes. Plain Parquet has no mechanism to track which rows changed, so UPDATE operations are impossible.
- First, register both file paths as SQL tables so you can run SQL against them. You also need to write a managed Delta table for the time travel demonstration later. Add a new cell and run:
# Register the Parquet files as a table (Parquet format)
spark.sql("DROP TABLE IF EXISTS workspace.default.orders_parquet")
orders_df.write.format("parquet").mode("overwrite").saveAsTable("workspace.default.orders_parquet")
# Register the Delta files as a managed Delta table
spark.sql("DROP TABLE IF EXISTS workspace.default.orders_delta")
orders_df.write.format("delta").mode("overwrite").saveAsTable("workspace.default.orders_delta")
print("Both tables registered in Unity Catalog!")Why register as managed tables?
The file-based writes earlier let you inspect raw file structures. Managed tables let you run SQL commands like UPDATE and time travel queries. Delta Lake is the default format in Databricks, so saveAsTable creates a Delta table unless you specify another format.
- Run the cell. You should see Both tables registered in Unity Catalog! in the output.
- Now try to UPDATE the Parquet table. Add a new cell and run:
# Attempt to UPDATE the Parquet table - this will FAIL
try:
spark.sql("""
UPDATE workspace.default.orders_parquet
SET order_status = 'processing'
WHERE order_status = 'created'
""")
except Exception as e:
print("UPDATE on Parquet table FAILED!")
print(f"Error: {e}")What should I see?
The UPDATE fails with an error. The Databricks documentation confirms that the UPDATE statement is only supported for Delta Lake tables. Parquet tables have no transaction log to record which rows changed, so row-level modifications are impossible.
- Run the cell. You should see the UPDATE on Parquet table FAILED! message followed by an error.
- Now run the same UPDATE on the Delta table. Add a new cell and run:
# UPDATE the Delta table - this SUCCEEDS
from delta.tables import DeltaTable
# Use the DeltaTable API to update rows
delta_table = DeltaTable.forName(spark, "workspace.default.orders_delta")
delta_table.update(
condition="order_status = 'created'",
set={"order_status": "'processing'"}
)
print("UPDATE on Delta table SUCCEEDED!")
# Verify the change
display(
spark.sql("SELECT order_status, COUNT(*) as count FROM workspace.default.orders_delta GROUP BY order_status ORDER BY count DESC")
)What does this code do?
DeltaTable.forName() creates a reference to the managed Delta table. The update() method changes every row where order_status equals created to processing instead.
This works because Delta's transaction log records the change as a new version. The old data files remain untouched. A new file is written with the updated rows, and the log tracks which files are current.
- Run the cell. You should see UPDATE on Delta table SUCCEEDED! followed by a table showing processing in the status counts (where created used to be).
Seeing an import error for DeltaTable?
Make sure the import line from delta.tables import DeltaTable is included in the same cell. In Databricks Free Edition, the delta library is pre-installed on serverless compute. If you still see an error, try restarting your session from the cluster menu.
Help me fix my DeltaTable import error in Databricks.
Demonstrate time travel on Delta vs Parquet
Time travel lets you query any historical version of a Delta table. Because Parquet has no transaction log, it has no concept of versions. You cannot go back in time with plain Parquet.
- Query the original version of the Delta table (before the UPDATE). Add a new cell and run:
# Time travel: read the original version (version 0) of the Delta table
original_df = spark.read.option("versionAsOf", 0).table("workspace.default.orders_delta")
print("=== Original data (version 0) - before the UPDATE ===")
display(
original_df.groupBy("order_status").count().orderBy("count", ascending=False)
)
# Compare with the current version
print("\n=== Current data (latest version) - after the UPDATE ===")
display(
spark.sql("SELECT order_status, COUNT(*) as count FROM workspace.default.orders_delta GROUP BY order_status ORDER BY count DESC")
)What does this code do?
spark.read.option("versionAsOf", 0).table(...) tells Spark to read the table as it was at version 0, which is the original write before any updates. You can compare the order_status counts between the original version and the current version.
In the original version, you will see created in the status list. In the current version, those rows now show processing instead.
- Run the cell. You should see two tables: one showing the original statuses (with created) and one showing the updated statuses (with processing replacing created).
- Now try time travel on the Parquet table to prove it is impossible. Add a new cell and run:
# Attempt time travel on the Parquet table - this will FAIL
try:
parquet_time_travel = spark.read.option("versionAsOf", 0).table("workspace.default.orders_parquet")
display(parquet_time_travel.limit(5))
except Exception as e:
print("Time travel on Parquet table FAILED!")
print(f"Error: {e}")
print("\nParquet has no transaction log, so there is no concept of 'versions' to travel back to.")Why does time travel fail on Parquet?
Time travel requires a transaction log that records every change as a numbered version. Parquet files are just static data. There is no log, no version numbers, and no way to reconstruct a previous state.
- Run the cell. You should see Time travel on Parquet table FAILED! along with an error message confirming that the operation is not supported.
- Finally, add a markdown summary cell to document your findings. Click the + button to add a new cell, then change the cell type to Markdown using the cell language dropdown. Paste this summary:
## Summary: Delta Lake vs Parquet
**Key takeaway:** Delta Lake = Parquet files + transaction log (`_delta_log`).
The transaction log enables:
- **ACID transactions** - writes are atomic; partial failures never corrupt the table
- **Time travel** - query any historical version by number or timestamp
- **Schema evolution** - safely add new columns without breaking existing queries
- **DML operations** - UPDATE, DELETE, and MERGE work at the row level
Without the transaction log, Parquet is just a static file format with no transactional guarantees.
| Feature | Parquet | Delta Lake |
|---------|---------|------------|
| Row-level UPDATE | Not supported | Supported |
| Row-level DELETE | Not supported | Supported |
| Time travel | Not possible | Query any version |
| ACID transactions | No | Yes |
| Schema evolution | Manual | Built-in |- Run the markdown cell (Shift+Enter) to render it. You should see a formatted summary with a comparison table.
Cell not rendering as markdown?
Make sure you changed the cell type before pasting. Click the language dropdown at the top of the cell (it shows Python by default) and change it to Markdown. Alternatively, type %md at the very first line of the cell to switch it to markdown mode.
Help me change a cell to markdown in Databricks.
✔️ Awesome, I've got everything!
Great. Double check you have saved your notebook. All 8 cells should have run successfully with the comparison results visible.
ⓧ I'd like to double check the full code
Here is the complete notebook code for all Python cells (the markdown cell is separate):
# Cell 1: Load orders data from the Unity Catalog volume
orders_df = spark.read.csv(
"/Volumes/workspace/default/olist-data/olist_orders_dataset.csv",
header=True,
inferSchema=True
)
# Preview the first few rows
display(orders_df.limit(5))# Cell 2: Write orders as plain Parquet files to the volume
parquet_path = "/Volumes/workspace/default/olist-data/orders_parquet"
orders_df.write.format("parquet").mode("overwrite").save(parquet_path)
# Write orders as Delta files to the volume
delta_path = "/Volumes/workspace/default/olist-data/orders_delta"
orders_df.write.format("delta").mode("overwrite").save(delta_path)
print("Both formats written successfully!")# Cell 3: List files in the Parquet directory
print("=== Parquet directory ===")
display(dbutils.fs.ls("/Volumes/workspace/default/olist-data/orders_parquet"))
print("\n=== Delta directory ===")
display(dbutils.fs.ls("/Volumes/workspace/default/olist-data/orders_delta"))# Cell 4: Look inside the _delta_log directory
display(dbutils.fs.ls("/Volumes/workspace/default/olist-data/orders_delta/_delta_log"))# Cell 5: Register both as managed tables for DML comparison
spark.sql("DROP TABLE IF EXISTS workspace.default.orders_parquet")
orders_df.write.format("parquet").mode("overwrite").saveAsTable("workspace.default.orders_parquet")
spark.sql("DROP TABLE IF EXISTS workspace.default.orders_delta")
orders_df.write.format("delta").mode("overwrite").saveAsTable("workspace.default.orders_delta")
print("Both tables registered in Unity Catalog!")# Cell 6: Attempt to UPDATE the Parquet table - this will FAIL
try:
spark.sql("""
UPDATE workspace.default.orders_parquet
SET order_status = 'processing'
WHERE order_status = 'created'
""")
except Exception as e:
print("UPDATE on Parquet table FAILED!")
print(f"Error: {e}")# Cell 7: UPDATE the Delta table - this SUCCEEDS
from delta.tables import DeltaTable
# Use the DeltaTable API to update rows
delta_table = DeltaTable.forName(spark, "workspace.default.orders_delta")
delta_table.update(
condition="order_status = 'created'",
set={"order_status": "'processing'"}
)
print("UPDATE on Delta table SUCCEEDED!")
# Verify the change
display(
spark.sql("SELECT order_status, COUNT(*) as count FROM workspace.default.orders_delta GROUP BY order_status ORDER BY count DESC")
)# Cell 8: Time travel - read the original version (version 0) of the Delta table
original_df = spark.read.option("versionAsOf", 0).table("workspace.default.orders_delta")
print("=== Original data (version 0) - before the UPDATE ===")
display(
original_df.groupBy("order_status").count().orderBy("count", ascending=False)
)
# Compare with the current version
print("\n=== Current data (latest version) - after the UPDATE ===")
display(
spark.sql("SELECT order_status, COUNT(*) as count FROM workspace.default.orders_delta GROUP BY order_status ORDER BY count DESC")
)# Cell 9: Attempt time travel on the Parquet table - this will FAIL
try:
parquet_time_travel = spark.read.option("versionAsOf", 0).table("workspace.default.orders_parquet")
display(parquet_time_travel.limit(5))
except Exception as e:
print("Time travel on Parquet table FAILED!")
print(f"Error: {e}")
print("\nParquet has no transaction log, so there is no concept of 'versions' to travel back to.")You have proven the fundamental difference between Delta Lake and Parquet through hands-on experiments. Next up, you will use these Delta Lake superpowers to build a proper Bronze layer that ingests all 8 CSV files with full ACID guarantees.
Build the Bronze Layer
In the previous step, you proved that Delta Lake's transaction log is what separates it from plain Parquet files. Now it's time to put that knowledge to work by building a real ingestion pipeline.
The Bronze layer is the first tier of a Medallion Architecture. It stores raw data exactly as it arrived from the source, with metadata tracking when and where it came from. Because every write is an ACID transaction, the Bronze layer guarantees that your raw data is never left in a half-written, corrupted state.
In this step, get ready to:
- Define a reusable ingestion function that reads CSVs and writes Delta tables with metadata columns.
- Ingest all 8 Olist CSV files into Bronze Delta tables and verify ACID guarantees.
- Confirm record counts and review the audit trail for every table.
Create the notebook and define the ingestion function
Before writing any data, you need a function that handles the repeatable work of reading a CSV, adding tracking columns, and saving it as a managed Delta Lake table.
- Click + New in the left sidebar of your Databricks workspace.
- Select Notebook from the menu.
- Click the notebook title at the top and rename it to 02_bronze_layer.
- In the first cell, paste the following ingestion function:
from pyspark.sql.functions import current_timestamp, lit
def ingest_to_bronze(filename, table_name):
"""Read a CSV from the volume and write it as a managed Delta table
with metadata columns for lineage tracking."""
# Read the CSV file from the Unity Catalog volume
df = spark.read.format("csv") \
.option("header", "true") \
.option("inferSchema", "true") \
.load(f"/Volumes/workspace/default/olist-data/{filename}")
# Add metadata columns: when it was ingested and which file it came from
df_with_metadata = df \
.withColumn("ingestion_timestamp", current_timestamp()) \
.withColumn("source_file", lit(filename))
# Write as a managed Delta table (Delta is the default format)
df_with_metadata.write.mode("overwrite").saveAsTable(f"workspace.default.{table_name}")
print(f"Ingested {filename} -> workspace.default.{table_name} ({df_with_metadata.count()} rows)")What does this function do?
The function handles three responsibilities for each CSV file:
- It reads the CSV from your Unity Catalog volume at /Volumes/workspace/default/olist-data/, inferring the schema automatically.
- It adds ingestion_timestamp (when the data was loaded) and source_file (which file it came from) as metadata columns for lineage.
- It writes the result as a managed Delta table using saveAsTable(). Because Delta is the default format in Databricks, this creates a Delta table with full transactional guarantees.
- Run the cell by pressing Shift+Enter.
- You should see the cell complete without errors. The function is now defined in memory and ready to use.
Seeing a NameError or import error?
Make sure the cell starts with the from pyspark.sql.functions import current_timestamp, lit line. If you see a compute-related error, wait a moment for serverless compute to start up and run the cell again.
Help me debug this import error in my Databricks notebook.
Ingest all tables and demonstrate ACID guarantees
With the function ready, you can now ingest all 8 Olist CSV files in one go. Each write is an atomic transaction. If anything goes wrong mid-write, the table stays at its last valid version.
- Add a new cell below and paste the following code to ingest all 8 tables:
# Define the mapping of CSV files to Bronze table names
tables = [
("olist_orders_dataset.csv", "bronze_orders"),
("olist_order_items_dataset.csv", "bronze_order_items"),
("olist_customers_dataset.csv", "bronze_customers"),
("olist_products_dataset.csv", "bronze_products"),
("olist_order_payments_dataset.csv", "bronze_order_payments"),
("olist_order_reviews_dataset.csv", "bronze_order_reviews"),
("olist_sellers_dataset.csv", "bronze_sellers"),
("product_category_name_translation.csv", "bronze_product_category_name_translation"),
]
# Ingest each CSV file into a Bronze Delta table
for filename, table_name in tables:
ingest_to_bronze(filename, table_name)
print("\nBronze layer ingestion complete!")What's happening here?
The tables list maps each CSV filename to the Delta table name it should become. The loop calls your ingestion function for each pair, creating 8 managed Delta tables in the workspace.default schema.
Each saveAsTable() call is a separate ACID transaction. If one table's write fails, the others remain intact. No partial or corrupted data can sneak in.
- Run the cell by pressing Shift+Enter.
- You should see 8 printed lines, each showing the filename, table name, and row count. This takes a minute or two on serverless compute.
Seeing a path not found error?
Double-check that your volume is named olist-data (with a hyphen) and that all 8 CSV files were uploaded in Step 1. Open Catalog in the sidebar, expand workspace > default > Volumes, and confirm the files are listed.
Help me fix the path not found error when reading CSVs from my Unity Catalog volume.
Now verify that the first write was committed atomically by checking the table history.
- Add a new cell and run the following to inspect the transaction history:
# Show the transaction history for bronze_orders
# Version 0 = the initial atomic write
display(spark.sql("DESCRIBE HISTORY workspace.default.bronze_orders"))- You should see a table with version 0 showing the WRITE operation, a timestamp, and operation metrics like the number of files and rows added.
Why does version 0 matter?
Version 0 proves that the entire table was created in a single atomic commit. There is no version 0.5 or partial state. The transaction log recorded every file added during the write, and only made the table visible to readers after all files were committed successfully.
Next, simulate what happens when a bad write occurs to see ACID protection in action.
- Add a new cell and paste the following ACID failure simulation:
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
# Create a DataFrame with a completely wrong schema
bad_schema = StructType([
StructField("wrong_column", IntegerType(), True),
StructField("bad_data", StringType(), True)
])
bad_data = spark.createDataFrame([(1, "corrupt"), (2, "invalid")], schema=bad_schema)
# Try to append bad data to bronze_orders (schema enforcement will reject it)
try:
bad_data.write.mode("append").saveAsTable("workspace.default.bronze_orders")
print("Write succeeded (unexpected!)")
except Exception as e:
print(f"Write REJECTED: {type(e).__name__}")
print(f"\nThe table is safe! Checking current row count...")
count = spark.table("workspace.default.bronze_orders").count()
print(f"bronze_orders still has {count} rows - no corruption.")What does this demonstrate?
Delta Lake's schema enforcement rejects the mismatched DataFrame before any data is written. The table stays at its last valid version with the correct row count.
In production, this means accidental writes with wrong column names, missing fields, or type mismatches are caught immediately. Your Bronze layer remains a trustworthy source of raw data.
- Run the cell. You should see the "Write REJECTED" message followed by the original row count confirming the table is unchanged.
The write succeeded instead of failing?
If the write mode was overwrite instead of append, Databricks may replace the table entirely. Make sure the code uses .mode("append") so that schema enforcement compares the new data against the existing table schema.
Help me understand why my ACID failure simulation didn't reject the bad write.
Verify the Bronze layer and review the audit trail
With all tables ingested, you need to confirm every table has data and inspect the full audit history.
- Add a new cell and run the following to check record counts across all 8 tables:
# Verify record counts for all Bronze tables
print("Bronze Layer Record Counts:")
print("=" * 50)
for filename, table_name in tables:
count = spark.table(f"workspace.default.{table_name}").count()
print(f" {table_name}: {count:,} rows")- You should see row counts for all 8 tables printed out. The bronze_orders table should show approximately 99,441 rows.
- Add another cell to display sample data from the orders table:
# Display sample data from bronze_orders to verify structure
# Notice the ingestion_timestamp and source_file columns at the end
display(spark.table("workspace.default.bronze_orders").limit(5))- Run the cell. You should see a formatted table with the original order columns plus your two metadata columns (ingestion_timestamp and source_file) on the right.
Finally, use the DeltaTable API to view the full audit trail programmatically.
- Add a final cell and run the following:
from delta.tables import DeltaTable
# Show the full audit trail using the DeltaTable API
delta_table = DeltaTable.forName(spark, "workspace.default.bronze_orders")
display(delta_table.history())What is the DeltaTable audit trail?
The DeltaTable.forName() method gives you a Python handle to any Delta table by its catalog path. Calling .history() returns provenance information for every write operation: who ran it, when, what operation type, and how many rows were affected.
This audit trail is automatic. You never have to set up logging or tracking manually. Every Delta table maintains its own complete history in the _delta_log directory.
- Run the cell. You should see the history table showing version 0 (and possibly version 1 if the failed append attempt created an entry) with timestamps and operation details.
What is the Bronze layer's purpose?
The Bronze layer stores raw data exactly as it arrived from the source. It is immutable and append-only by convention. You never modify Bronze data directly.
This means you always have a full history of what arrived and when. If transformations in later layers introduce bugs, you can always re-process from Bronze. The metadata columns (ingestion_timestamp and source_file) make it easy to trace any record back to its origin.
Missing tables or unexpected counts?
Open Catalog in the sidebar and navigate to workspace > default > Tables. You should see all 8 bronze_ tables listed. If any are missing, re-run the ingestion cell and check for error messages in its output.
Help me figure out why some of my Bronze tables are missing or have wrong row counts.
✔️ Awesome, I've got everything!
Great. Double check you've saved your notebook by pressing Cmd+S (macOS) or Ctrl+S (Windows).
ⓧ I'd like to double check the full code
Here is the complete 02_bronze_layer notebook with all cells combined:
from pyspark.sql.functions import current_timestamp, lit
def ingest_to_bronze(filename, table_name):
"""Read a CSV from the volume and write it as a managed Delta table
with metadata columns for lineage tracking."""
# Read the CSV file from the Unity Catalog volume
df = spark.read.format("csv") \
.option("header", "true") \
.option("inferSchema", "true") \
.load(f"/Volumes/workspace/default/olist-data/{filename}")
# Add metadata columns: when it was ingested and which file it came from
df_with_metadata = df \
.withColumn("ingestion_timestamp", current_timestamp()) \
.withColumn("source_file", lit(filename))
# Write as a managed Delta table (Delta is the default format)
df_with_metadata.write.mode("overwrite").saveAsTable(f"workspace.default.{table_name}")
print(f"Ingested {filename} -> workspace.default.{table_name} ({df_with_metadata.count()} rows)")
# ------- Cell 2 -------
# Define the mapping of CSV files to Bronze table names
tables = [
("olist_orders_dataset.csv", "bronze_orders"),
("olist_order_items_dataset.csv", "bronze_order_items"),
("olist_customers_dataset.csv", "bronze_customers"),
("olist_products_dataset.csv", "bronze_products"),
("olist_order_payments_dataset.csv", "bronze_order_payments"),
("olist_order_reviews_dataset.csv", "bronze_order_reviews"),
("olist_sellers_dataset.csv", "bronze_sellers"),
("product_category_name_translation.csv", "bronze_product_category_name_translation"),
]
# Ingest each CSV file into a Bronze Delta table
for filename, table_name in tables:
ingest_to_bronze(filename, table_name)
print("\nBronze layer ingestion complete!")
# ------- Cell 3 -------
# Show the transaction history for bronze_orders
# Version 0 = the initial atomic write
display(spark.sql("DESCRIBE HISTORY workspace.default.bronze_orders"))
# ------- Cell 4 -------
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
# Create a DataFrame with a completely wrong schema
bad_schema = StructType([
StructField("wrong_column", IntegerType(), True),
StructField("bad_data", StringType(), True)
])
bad_data = spark.createDataFrame([(1, "corrupt"), (2, "invalid")], schema=bad_schema)
# Try to append bad data to bronze_orders (schema enforcement will reject it)
try:
bad_data.write.mode("append").saveAsTable("workspace.default.bronze_orders")
print("Write succeeded (unexpected!)")
except Exception as e:
print(f"Write REJECTED: {type(e).__name__}")
print(f"\nThe table is safe! Checking current row count...")
count = spark.table("workspace.default.bronze_orders").count()
print(f"bronze_orders still has {count} rows - no corruption.")
# ------- Cell 5 -------
# Verify record counts for all Bronze tables
print("Bronze Layer Record Counts:")
print("=" * 50)
for filename, table_name in tables:
count = spark.table(f"workspace.default.{table_name}").count()
print(f" {table_name}: {count:,} rows")
# ------- Cell 6 -------
# Display sample data from bronze_orders to verify structure
# Notice the ingestion_timestamp and source_file columns at the end
display(spark.table("workspace.default.bronze_orders").limit(5))
# ------- Cell 7 -------
from delta.tables import DeltaTable
# Show the full audit trail using the DeltaTable API
delta_table = DeltaTable.forName(spark, "workspace.default.bronze_orders")
display(delta_table.history())Your Bronze layer is locked in. You have 8 raw Delta tables with full ACID protection and automatic audit trails. Next up, you'll clean and transform this raw data into polished Silver layer tables using MERGE operations and deduplication.
Build the Silver Layer
Your Bronze layer now holds all the raw Olist data exactly as it came from the CSV files. Raw data is messy though. Date columns stored as strings can't be compared or filtered properly. Text fields have inconsistent casing and trailing whitespace. Some rows are outright duplicates.
The Silver layer solves these problems. In this step, you'll read from Bronze, apply cleaning transformations, remove duplicates, and demonstrate a MERGE (upsert) operation that updates existing records while inserting new ones.
In this step, get ready to:
- Clean and deduplicate Bronze data into Silver Delta tables.
- Demonstrate MERGE (upsert) operations on the Silver customers table.
- Build an enriched joined table combining orders, customers, items, products, and payments.
Create the Silver notebook and clean the orders table
The Silver layer transforms raw Bronze data into analysis-ready tables. Each table gets the same treatment: cast date strings to proper timestamps, trim and lowercase text fields, handle nulls, and drop duplicates.
- In your Databricks workspace, click New in the sidebar, then select Notebook.
- Name the notebook 03_silver_layer and make sure Python is selected as the default language.
The orders table is a good starting point because it has multiple date columns that need timestamp casting.
- Add the following code to the first cell of your notebook to import your dependencies and clean the orders table:
from pyspark.sql.functions import col, trim, lower, to_timestamp, current_timestamp
from delta.tables import DeltaTable
# Read raw orders from the Bronze layer
df_orders = spark.table("workspace.default.bronze_orders")
# Cast date string columns to proper timestamp types
df_orders_clean = df_orders \
.withColumn("order_purchase_timestamp", to_timestamp(col("order_purchase_timestamp"))) \
.withColumn("order_approved_at", to_timestamp(col("order_approved_at"))) \
.withColumn("order_delivered_carrier_date", to_timestamp(col("order_delivered_carrier_date"))) \
.withColumn("order_delivered_customer_date", to_timestamp(col("order_delivered_customer_date"))) \
.withColumn("order_estimated_delivery_date", to_timestamp(col("order_estimated_delivery_date"))) \
.withColumn("order_status", lower(trim(col("order_status")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["order_id"])
# Write cleaned orders as a Silver Delta table
df_orders_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_orders")
print(f"Silver orders count: {spark.table('workspace.default.silver_orders').count()}")
display(spark.table("workspace.default.silver_orders").limit(5))What does this code do?
- to_timestamp() converts string date columns into proper timestamp types so you can filter and compare dates correctly.
- lower(trim(col(...))) removes leading/trailing whitespace and lowercases text to standardize values.
- .drop("ingestion_timestamp", "source_file") removes the Bronze metadata columns since they're only needed for auditing the raw layer.
- .dropDuplicates(["order_id"]) keeps only one row per unique order ID, eliminating any duplicate records.
- Run the cell. You should see a count printed (approximately 99,441 orders) and a sample of 5 rows displayed with properly typed timestamp columns.
Seeing a table already exists error?
If you re-run the cell after the table was already created, the overwrite mode should handle it cleanly. If you see an error about the table existing in a different format, drop it first with spark.sql("DROP TABLE IF EXISTS workspace.default.silver_orders") and re-run.
help me fix a table already exists error in Databricks
Now apply the same cleaning pattern to all the remaining Bronze tables.
- Add a new cell with the following code to clean customers, order items, products, payments, reviews, and sellers:
# --- Clean customers ---
df_customers = spark.table("workspace.default.bronze_customers")
df_customers_clean = df_customers \
.withColumn("customer_city", lower(trim(col("customer_city")))) \
.withColumn("customer_state", lower(trim(col("customer_state")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["customer_id"])
df_customers_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_customers")
# --- Clean order items ---
df_items = spark.table("workspace.default.bronze_order_items")
df_items_clean = df_items \
.withColumn("shipping_limit_date", to_timestamp(col("shipping_limit_date"))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["order_id", "order_item_id"])
df_items_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_order_items")
# --- Clean products ---
df_products = spark.table("workspace.default.bronze_products")
df_products_clean = df_products \
.withColumn("product_category_name", lower(trim(col("product_category_name")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["product_id"])
df_products_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_products")
# --- Clean payments ---
df_payments = spark.table("workspace.default.bronze_order_payments")
df_payments_clean = df_payments \
.withColumn("payment_type", lower(trim(col("payment_type")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates()
df_payments_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_order_payments")
# --- Clean reviews ---
df_reviews = spark.table("workspace.default.bronze_order_reviews")
df_reviews_clean = df_reviews \
.withColumn("review_creation_date", to_timestamp(col("review_creation_date"))) \
.withColumn("review_answer_timestamp", to_timestamp(col("review_answer_timestamp"))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["review_id"])
df_reviews_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_order_reviews")
# --- Clean sellers ---
df_sellers = spark.table("workspace.default.bronze_sellers")
df_sellers_clean = df_sellers \
.withColumn("seller_city", lower(trim(col("seller_city")))) \
.withColumn("seller_state", lower(trim(col("seller_state")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["seller_id"])
df_sellers_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_sellers")
print("All Silver tables created successfully!")
print(f" silver_customers: {spark.table('workspace.default.silver_customers').count()}")
print(f" silver_order_items: {spark.table('workspace.default.silver_order_items').count()}")
print(f" silver_products: {spark.table('workspace.default.silver_products').count()}")
print(f" silver_order_payments: {spark.table('workspace.default.silver_order_payments').count()}")
print(f" silver_order_reviews: {spark.table('workspace.default.silver_order_reviews').count()}")
print(f" silver_sellers: {spark.table('workspace.default.silver_sellers').count()}")Why the same pattern for every table?
Each table gets the same three-step treatment: cast dates to timestamps, standardize text with lower(trim(...)), and remove duplicates using a primary key column. This consistency makes the Silver layer predictable.
The metadata columns (ingestion_timestamp and source_file) stay in Bronze for auditing purposes. Silver only carries business data.
- Run the cell. You should see row counts printed for all six Silver tables, confirming each one was created successfully.
Cell taking a long time?
Databricks Free Edition uses serverless compute with daily quota limits. If you hit the quota, compute pauses until the next day. Your data is preserved.
help me understand Databricks Free Edition compute quotas
Demonstrate MERGE (upsert) on the Silver customers table
In production, data arrives in batches. Some records are new, some are updates to existing rows. A MERGE operation handles both in a single atomic transaction: it updates rows that already exist and inserts rows that are new.
- Add a new cell to simulate an update batch and run a MERGE against the Silver customers table:
# Simulate an update batch: take 10 customers and change their city
existing_customers = spark.table("workspace.default.silver_customers").limit(10)
# Create a simulated update: change the city for these customers
updates_df = existing_customers.withColumn("customer_city", lower(trim(col("customer_city"))))
updates_df = updates_df.withColumn("customer_city",
col("customer_city").substr(1, 3) # Truncate city to simulate a change
)
print(f"Update batch size: {updates_df.count()} records")
display(updates_df.select("customer_id", "customer_city").limit(5))What is this simulating?
In a real pipeline, an upstream system might send corrected addresses or updated customer details. Here you're simulating that scenario by modifying the city field for 10 existing customers. The MERGE in the next cell handles the update atomically.
- Run the cell to create your simulated update batch. You should see 10 records displayed with shortened city names.
- Add a new cell to execute the MERGE operation:
# Perform MERGE (upsert) on the Silver customers table
deltaTable = DeltaTable.forName(spark, "workspace.default.silver_customers")
deltaTable.alias("target").merge(
updates_df.alias("source"),
"target.customer_id = source.customer_id"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
print("MERGE complete! Checking table history...")
# Show the version history to see the MERGE operation recorded
display(deltaTable.history())What does this MERGE do?
- DeltaTable.forName(spark, "workspace.default.silver_customers") gets a reference to the existing Delta table.
- .alias("target").merge(updates_df.alias("source"), ...) sets up the join condition between existing data and the incoming batch.
- .whenMatchedUpdateAll() updates all columns when a matching customer_id is found.
- .whenNotMatchedInsertAll() inserts the full row if no matching customer_id exists in the target.
The history output shows a new version created with operation MERGE. This is the ACID guarantee in action. The update either succeeds completely or not at all.
- Run the cell. You should see "MERGE complete!" followed by a history table showing version 0 (the initial write) and version 1 (the MERGE operation).
Seeing an AnalysisException?
If you see an error about the table not being found, make sure you ran the previous cells that created workspace.default.silver_customers first. Delta MERGE requires the target table to already exist.
help me debug a DeltaTable.forName error in Databricks
Build the enriched orders table
Individual Silver tables are clean, but analytics often require data from multiple tables combined. The enriched table joins orders with customers, items, products, and payments to create a single wide table ready for Gold layer aggregations.
- Add a new cell to build the silver_orders_enriched table by joining your Silver tables together:
# Read Silver tables for joining
orders = spark.table("workspace.default.silver_orders")
customers = spark.table("workspace.default.silver_customers")
order_items = spark.table("workspace.default.silver_order_items")
products = spark.table("workspace.default.silver_products")
payments = spark.table("workspace.default.silver_order_payments")
# Join orders with customers, items, products, and payments
df_enriched = orders \
.join(customers, orders.customer_id == customers.customer_id, "left") \
.join(order_items, orders.order_id == order_items.order_id, "left") \
.join(products, order_items.product_id == products.product_id, "left") \
.join(payments, orders.order_id == payments.order_id, "left") \
.select(
orders.order_id,
orders.order_status,
orders.order_purchase_timestamp,
customers.customer_city,
customers.customer_state,
products.product_category_name,
order_items.price,
order_items.freight_value,
payments.payment_type,
payments.payment_value
)
# Write the enriched table
df_enriched.write.mode("overwrite").saveAsTable("workspace.default.silver_orders_enriched")
print(f"Enriched table count: {spark.table('workspace.default.silver_orders_enriched').count()}")
display(spark.table("workspace.default.silver_orders_enriched").limit(10))What does this join produce?
The enriched table combines order details with customer location, product category, and payment information into one flat table. Each row represents an order line item with its full context.
Left joins ensure you keep all orders even if a matching customer or product record is missing. Missing fields show up as null rather than dropping the order entirely.
- Run the cell. You should see a row count and 10 sample rows showing columns like order_id, customer_city, product_category_name, and payment_type all in one table.
Verify the enriched data makes sense by running a quick analytical query.
- Add a new cell with this verification query:
# Verify: show order details with customer city, product category, and payment method
display(
spark.sql("""
SELECT
order_id,
customer_city,
product_category_name,
payment_type,
payment_value
FROM workspace.default.silver_orders_enriched
WHERE product_category_name IS NOT NULL
LIMIT 20
""")
)- Run the cell. You should see 20 rows with real city names, product categories, and payment types confirming the join worked correctly.
Seeing all nulls in your results?
If product categories or customer cities are all null, the join keys may not be matching. Check that your Silver tables contain data by running spark.table("workspace.default.silver_customers").count() to confirm they're not empty.
help me debug null values in my joined Delta table
✔️ Awesome, I've got everything!
Great. Make sure all cells in your 03_silver_layer notebook have run successfully.
ⓧ I'd like to double check the full code
Here's the complete 03_silver_layer notebook for reference. Each cell is separated by a comment divider:
# Cell 1: Clean orders table
from pyspark.sql.functions import col, trim, lower, to_timestamp, current_timestamp
from delta.tables import DeltaTable
# Read raw orders from the Bronze layer
df_orders = spark.table("workspace.default.bronze_orders")
# Cast date string columns to proper timestamp types
df_orders_clean = df_orders \
.withColumn("order_purchase_timestamp", to_timestamp(col("order_purchase_timestamp"))) \
.withColumn("order_approved_at", to_timestamp(col("order_approved_at"))) \
.withColumn("order_delivered_carrier_date", to_timestamp(col("order_delivered_carrier_date"))) \
.withColumn("order_delivered_customer_date", to_timestamp(col("order_delivered_customer_date"))) \
.withColumn("order_estimated_delivery_date", to_timestamp(col("order_estimated_delivery_date"))) \
.withColumn("order_status", lower(trim(col("order_status")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["order_id"])
# Write cleaned orders as a Silver Delta table
df_orders_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_orders")
print(f"Silver orders count: {spark.table('workspace.default.silver_orders').count()}")
display(spark.table("workspace.default.silver_orders").limit(5))
# -------------------------------------------------------------------
# Cell 2: Clean all remaining Bronze tables
# --- Clean customers ---
df_customers = spark.table("workspace.default.bronze_customers")
df_customers_clean = df_customers \
.withColumn("customer_city", lower(trim(col("customer_city")))) \
.withColumn("customer_state", lower(trim(col("customer_state")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["customer_id"])
df_customers_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_customers")
# --- Clean order items ---
df_items = spark.table("workspace.default.bronze_order_items")
df_items_clean = df_items \
.withColumn("shipping_limit_date", to_timestamp(col("shipping_limit_date"))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["order_id", "order_item_id"])
df_items_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_order_items")
# --- Clean products ---
df_products = spark.table("workspace.default.bronze_products")
df_products_clean = df_products \
.withColumn("product_category_name", lower(trim(col("product_category_name")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["product_id"])
df_products_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_products")
# --- Clean payments ---
df_payments = spark.table("workspace.default.bronze_order_payments")
df_payments_clean = df_payments \
.withColumn("payment_type", lower(trim(col("payment_type")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates()
df_payments_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_order_payments")
# --- Clean reviews ---
df_reviews = spark.table("workspace.default.bronze_order_reviews")
df_reviews_clean = df_reviews \
.withColumn("review_creation_date", to_timestamp(col("review_creation_date"))) \
.withColumn("review_answer_timestamp", to_timestamp(col("review_answer_timestamp"))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["review_id"])
df_reviews_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_order_reviews")
# --- Clean sellers ---
df_sellers = spark.table("workspace.default.bronze_sellers")
df_sellers_clean = df_sellers \
.withColumn("seller_city", lower(trim(col("seller_city")))) \
.withColumn("seller_state", lower(trim(col("seller_state")))) \
.drop("ingestion_timestamp", "source_file") \
.dropDuplicates(["seller_id"])
df_sellers_clean.write.mode("overwrite").saveAsTable("workspace.default.silver_sellers")
print("All Silver tables created successfully!")
print(f" silver_customers: {spark.table('workspace.default.silver_customers').count()}")
print(f" silver_order_items: {spark.table('workspace.default.silver_order_items').count()}")
print(f" silver_products: {spark.table('workspace.default.silver_products').count()}")
print(f" silver_order_payments: {spark.table('workspace.default.silver_order_payments').count()}")
print(f" silver_order_reviews: {spark.table('workspace.default.silver_order_reviews').count()}")
print(f" silver_sellers: {spark.table('workspace.default.silver_sellers').count()}")
# -------------------------------------------------------------------
# Cell 3: Simulate an update batch
# Simulate an update batch: take 10 customers and change their city
existing_customers = spark.table("workspace.default.silver_customers").limit(10)
# Create a simulated update: change the city for these customers
updates_df = existing_customers.withColumn("customer_city", lower(trim(col("customer_city"))))
updates_df = updates_df.withColumn("customer_city",
col("customer_city").substr(1, 3) # Truncate city to simulate a change
)
print(f"Update batch size: {updates_df.count()} records")
display(updates_df.select("customer_id", "customer_city").limit(5))
# -------------------------------------------------------------------
# Cell 4: Execute MERGE (upsert)
# Perform MERGE (upsert) on the Silver customers table
deltaTable = DeltaTable.forName(spark, "workspace.default.silver_customers")
deltaTable.alias("target").merge(
updates_df.alias("source"),
"target.customer_id = source.customer_id"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
print("MERGE complete! Checking table history...")
# Show the version history to see the MERGE operation recorded
display(deltaTable.history())
# -------------------------------------------------------------------
# Cell 5: Build enriched orders table
# Read Silver tables for joining
orders = spark.table("workspace.default.silver_orders")
customers = spark.table("workspace.default.silver_customers")
order_items = spark.table("workspace.default.silver_order_items")
products = spark.table("workspace.default.silver_products")
payments = spark.table("workspace.default.silver_order_payments")
# Join orders with customers, items, products, and payments
df_enriched = orders \
.join(customers, orders.customer_id == customers.customer_id, "left") \
.join(order_items, orders.order_id == order_items.order_id, "left") \
.join(products, order_items.product_id == products.product_id, "left") \
.join(payments, orders.order_id == payments.order_id, "left") \
.select(
orders.order_id,
orders.order_status,
orders.order_purchase_timestamp,
customers.customer_city,
customers.customer_state,
products.product_category_name,
order_items.price,
order_items.freight_value,
payments.payment_type,
payments.payment_value
)
# Write the enriched table
df_enriched.write.mode("overwrite").saveAsTable("workspace.default.silver_orders_enriched")
print(f"Enriched table count: {spark.table('workspace.default.silver_orders_enriched').count()}")
display(spark.table("workspace.default.silver_orders_enriched").limit(10))
# -------------------------------------------------------------------
# Cell 6: Verify enriched data
# Verify: show order details with customer city, product category, and payment method
display(
spark.sql("""
SELECT
order_id,
customer_city,
product_category_name,
payment_type,
payment_value
FROM workspace.default.silver_orders_enriched
WHERE product_category_name IS NOT NULL
LIMIT 20
""")
)Your Silver layer is complete. You have clean, deduplicated tables with proper data types and a MERGE operation tracked in the table history. Next, you'll use Delta Lake's time travel feature to query historical versions and recover from simulated data disasters.
Explore Time Travel and Data Recovery
In production, things go wrong. Someone runs a bad DELETE, a deployment corrupts data, or a regulator asks what your table looked like last Tuesday. Delta Lake's transaction log records every version of your table, letting you query any past state and recover from disasters.
In this step, you'll explore time travel by querying historical versions of your Silver tables. Then you'll simulate a data disaster and recover from it using RESTORE.
In this step, get ready to:
- Query historical table versions using Python and SQL time travel syntax.
- Simulate a data disaster and restore a table to a previous version.
- Understand VACUUM and its relationship to time travel retention.
Query historical versions with Python and SQL
Every write to a Delta Lake table creates a new version. Time travel lets you read any of those previous versions by specifying a version number or timestamp.
- In your Databricks workspace, click New in the sidebar, then select Notebook.
- Name the notebook 04_time_travel and keep the default language as Python.
- In the first cell, view the version history of silver_orders by running this code:
# View the full version history of silver_orders
from delta.tables import DeltaTable
deltaTable = DeltaTable.forName(spark, "workspace.default.silver_orders")
display(deltaTable.history())What does this code do?
DeltaTable.forName(spark, "workspace.default.silver_orders") creates a reference to your Silver orders Delta table.
The history() method returns provenance information for every write operation. You'll see columns like version, timestamp, operation, and operationMetrics. Results appear in reverse chronological order.
- Run the cell. You should see a table showing each version's operation (e.g., WRITE or MERGE) with timestamps.
- Note the version numbers in the output. Version 0 is the original write from your Silver layer notebook.
- In a new cell, query version 0 of the table using the Python time travel syntax:
# Time travel: read version 0 (the original Silver write)
df_v0 = spark.read.option('versionAsOf', 0).table("workspace.default.silver_orders")
print(f"Version 0 row count: {df_v0.count()}")
display(df_v0.limit(5))What does this code do?
The .option('versionAsOf', 0) option tells Spark to read the table as it existed at version 0, regardless of any changes made since then. This is time travel in action.
- Run the cell. You should see the row count and a sample of records from the original Silver layer write.
- In a new cell, query the same version using SQL syntax:
-- Time travel using SQL: query version 0 of silver_orders
SELECT * FROM workspace.default.silver_orders VERSION AS OF 0
LIMIT 5What is the SQL time travel syntax?
SQL uses VERSION AS OF directly after the table name. You can also use TIMESTAMP AS OF with a timestamp expression to query by time instead of version number.
- Run the SQL cell. You should see the same version 0 data you saw from the Python query.
- In a new cell, add a markdown heading documenting what each version represents:
# Display history alongside current count to show how versions map to operations
current_count = spark.table("workspace.default.silver_orders").count()
v0_count = spark.read.option('versionAsOf', 0).table("workspace.default.silver_orders").count()
print(f"Current version row count: {current_count}")
print(f"Version 0 row count: {v0_count}")- Run the cell. Both counts should match since no modifications have been made to silver_orders beyond the initial write.
Not seeing version history?
Make sure you ran the 03_silver_layer notebook successfully before starting this notebook. The Silver tables must exist with at least version 0.
If you see a table-not-found error, check that the table name matches exactly: workspace.default.silver_orders.
Help me debug my time travel query
Simulate a data disaster and restore with RESTORE
Time travel is not just for curiosity. It is a safety net for production systems. You are about to intentionally destroy all data in a table, then bring it back in one command.
- First, check the current state of silver_customers by running this cell:
# Check the current state of silver_customers before the disaster
from delta.tables import DeltaTable
customers_table = DeltaTable.forName(spark, "workspace.default.silver_customers")
print(f"Current row count: {spark.table('workspace.default.silver_customers').count()}")
# Show version history so we know which version to restore to
display(customers_table.history())Why silver_customers?
This table already has multiple versions from the MERGE upsert operations you ran in the previous step. That makes it a perfect candidate for demonstrating both the disaster and the recovery.
- Run the cell. Note the row count and the latest version number in the history output. You will need the latest version number to restore after the disaster.
- Now simulate the disaster. In a new cell, delete all records from the table:
-- DISASTER: accidentally delete ALL records from silver_customers
DELETE FROM workspace.default.silver_customersThis is intentional
You are deliberately wiping the table to demonstrate disaster recovery. In production, this might happen from a bad script, a misconfigured job, or an accidental WHERE clause omission.
- Run the DELETE cell.
- Verify the table is empty by running this cell:
# Confirm the disaster: table is now empty
empty_count = spark.table("workspace.default.silver_customers").count()
print(f"Row count after DELETE: {empty_count}")- Run the cell. You should see a row count of 0. The table is completely empty.
- Now restore the table using the RESTORE command. Replace the version number with the version you noted earlier (the last version before the DELETE):
-- RECOVERY: restore silver_customers to the version before the DELETE
RESTORE TABLE workspace.default.silver_customers TO VERSION AS OF 2How does RESTORE work?
RESTORE does not roll back the clock. It creates a new version of the table that has the same content as the target version. Your history still shows the DELETE and the RESTORE as separate events. Nothing is erased from the audit trail.
Use DESCRIBE HISTORY to find the version number you want to restore to. In your case, the version immediately before the DELETE is the target.
- Run the RESTORE cell. Databricks displays metrics including the number of restored files.
- Verify the data is back by running this cell:
# Verify the restore worked
restored_count = spark.table("workspace.default.silver_customers").count()
print(f"Row count after RESTORE: {restored_count}")
# Show the updated history including DELETE and RESTORE operations
customers_table = DeltaTable.forName(spark, "workspace.default.silver_customers")
display(customers_table.history())- Run the cell. You should see the original row count restored. The history now shows the RESTORE operation as the latest version.
When is RESTORE invaluable in production?
Accidental deletes from a bad WHERE clause or missing filter. Bad deployments that corrupt data in Silver or Gold layers. Regulatory audits that require reverting to a specific point in time.
Without Delta Lake, recovering from these scenarios requires restoring from backup systems. With RESTORE, recovery is a single SQL command that takes seconds.
RESTORE failed or returned unexpected counts?
Double check the version number you used. Run DESCRIBE HISTORY workspace.default.silver_customers to see all versions and pick the one before the DELETE operation.
Help me figure out which version to restore to
Understand VACUUM and its impact on time travel
Time travel is powerful, but it is not free. Every version of your table keeps old data files on disk. VACUUM is the cleanup command that removes those old files after a retention period.
- In a new cell, view the history of silver_customers to see all the versions that exist:
-- View the full history including our DELETE and RESTORE
DESCRIBE HISTORY workspace.default.silver_customers- Run the cell. You should see versions for the original WRITE, the MERGE, the DELETE, and the RESTORE.
- In a new cell, demonstrate VACUUM on the customers table:
# Demonstrate VACUUM (dry run - shows what would be cleaned up)
# VACUUM removes old data files beyond the retention period
customers_table = DeltaTable.forName(spark, "workspace.default.silver_customers")
customers_table.vacuum()What does VACUUM do?
VACUUM removes data files that are no longer referenced by the current table version and are older than the retention threshold. The default retention period is 7 days (controlled by delta.deletedFileRetentionDuration).
After VACUUM runs, you lose the ability to time travel to versions older than the retention period. The history metadata (from DESCRIBE HISTORY) remains visible for 30 days (controlled by delta.logRetentionDuration), but the underlying data files may be gone.
- Run the cell. Since your table was just created today, VACUUM will likely report that no files were deleted (they are all within the 7-day window).
- In a new cell, add a markdown explanation documenting the VACUUM trade-offs:
# Document the key VACUUM configuration properties
# delta.deletedFileRetentionDuration = "interval 7 days" (default)
# - Controls how long old data files are kept for time travel
# - Shorter = less storage cost, but less time travel depth
# - Longer = more storage cost, but more recovery options
#
# delta.logRetentionDuration = "interval 30 days" (default)
# - Controls how long the transaction log history is kept
# - This is the metadata (DESCRIBE HISTORY), not the data files
#
# The trade-off: storage cost vs. time travel depth
# In production, set retention based on your recovery SLA
# Example: To access 30 days of historical data:
# ALTER TABLE workspace.default.silver_customers
# SET TBLPROPERTIES (delta.deletedFileRetentionDuration = 'interval 30 days')
print("VACUUM configuration summary:")
print("- delta.deletedFileRetentionDuration (default: 7 days) = how long data files are kept")
print("- delta.logRetentionDuration (default: 30 days) = how long history metadata is kept")
print("- Trade-off: more retention = more storage cost, less retention = less time travel")The storage vs. safety trade-off
In production, you choose retention based on your recovery needs. A table with a 7-day retention can time travel up to 7 days back. A table with 30 days of retention costs more storage but gives you a full month of recovery window.
Databricks recommends setting a retention interval of at least 7 days. If you have long-running jobs, a shorter retention could cause VACUUM to delete files those jobs still need.
- Run the cell to confirm the summary prints without error.
Getting a retention check error?
If you try to VACUUM with a retention shorter than 7 days, Delta Lake has a safety check that prevents it. This protects you from accidentally deleting files that are still needed.
Help me understand the VACUUM retention error
Your time travel notebook is complete. You have demonstrated querying historical versions, recovering from a data disaster, and understanding the storage trade-offs of VACUUM. Next up, you will explore schema evolution to handle changing data structures gracefully.
Demonstrate Schema Evolution
In production, data structures change constantly. Upstream systems add new fields, rename columns, or restructure payloads. Your lakehouse needs to handle these changes gracefully without breaking existing queries or requiring full table rewrites.
Delta Lake solves this with two complementary features: schema enforcement (which rejects unexpected changes by default) and schema evolution (which safely integrates new columns when you opt in). In this step, you'll see both in action on your silver_customers table.
In this step, get ready to:
- Trigger a schema enforcement error by appending data with an extra column.
- Enable schema evolution with .option("mergeSchema", "true") to safely add new columns.
- Simulate a complex real-world schema change with multiple new fields arriving from an upstream system.
Create the schema evolution notebook
Schema enforcement is Delta Lake's default safety net. It prevents accidental data corruption by rejecting writes that don't match the table's existing schema.
- In your Databricks workspace, click New in the sidebar, then select Notebook.
- Name it 05_schema_evolution and make sure the language is set to Python.
First, you'll try to append a DataFrame that has a column the table doesn't know about. This simulates what happens when an upstream system adds a new field without warning.
- Add the following code to your first cell to read the current Silver customers table and create a sample batch with an extra column:
# Read the current silver_customers table
silver_customers = spark.table("workspace.default.silver_customers")
# Create a small sample batch with an extra column (loyalty_tier)
# This simulates an upstream system adding a new field
sample_with_new_col = (
silver_customers
.limit(10)
.withColumn("loyalty_tier", lit("gold"))
)
print(f"Current table columns: {silver_customers.columns}")
print(f"New batch columns: {sample_with_new_col.columns}")What does this code do?
You're reading the existing silver_customers table and creating a small 10-row sample that has one extra column called loyalty_tier. The print statements let you compare the column lists side by side.
- Run the cell. You should see the current table's column list followed by the new batch's list, which includes the extra loyalty_tier column at the end.
- Add a new cell and attempt to append this mismatched DataFrame without enabling schema evolution:
# Try to append data with an extra column (this will FAIL)
# Delta Lake's schema enforcement rejects unexpected columns
try:
sample_with_new_col.write.mode("append").saveAsTable("workspace.default.silver_customers")
print("Write succeeded (unexpected!)")
except Exception as e:
print("Schema enforcement blocked the write!")
print(f"Error: {e}")What does this code do?
The try/except block attempts the append and catches the error so the notebook keeps running. Delta Lake's schema enforcement detects that loyalty_tier doesn't exist in the target table and raises a DELTA_METADATA_MISMATCH error with a SCHEMA_MISMATCH subtype.
This is intentional. Without this guard, a typo in a column name or a rogue upstream change could silently corrupt your table. Schema enforcement is the default safety net that keeps your data consistent.
- Run the cell. You should see the message "Schema enforcement blocked the write!" followed by an error mentioning a schema mismatch.
Not seeing an error?
If the write succeeded without an error, your table may already have the loyalty_tier column from a previous run. Drop the table with spark.sql("DROP TABLE workspace.default.silver_customers") and re-run the Bronze and Silver layer notebooks to rebuild it.
help me understand why my schema enforcement test didn't produce an error
- Add a markdown cell documenting why schema enforcement matters:
# %md
# ## Schema Enforcement (Default Behavior)
# Delta Lake **rejects** writes that don't match the table's schema by default.
# This prevents:
# - Accidental column additions from upstream changes
# - Typos in column names silently creating new columns
# - Data type mismatches corrupting existing data
#
# The error `DELTA_METADATA_MISMATCH [SCHEMA_MISMATCH]` tells you exactly what went wrong
# and suggests using `.option("mergeSchema", "true")` to allow the change intentionally.Enable schema evolution to add a new column
Now that you've seen the guardrail in action, you'll intentionally opt in to schema evolution. This is what you do when a new column is a legitimate addition rather than a mistake.
- Add a new cell and re-run the append with .option("mergeSchema", "true") enabled:
# Enable schema evolution and re-run the append
# .option("mergeSchema", "true") tells Delta to add new columns automatically
sample_with_new_col.write.mode("append").option("mergeSchema", "true").saveAsTable("workspace.default.silver_customers")
print("Write succeeded with schema evolution enabled!")What does this code do?
The .option("mergeSchema", "true") flag tells Delta Lake to accept new columns from the incoming DataFrame. Delta adds the loyalty_tier column to the table's schema. All existing rows get NULL for this new column. Only the 10 new rows have the value gold populated.
- Run the cell. You should see "Write succeeded with schema evolution enabled!" with no errors.
- Add a new cell to verify the schema change and inspect the data:
# Verify the new column was added to the table schema
spark.sql("DESCRIBE TABLE workspace.default.silver_customers").show(truncate=False)
# Check existing rows (should have NULL for loyalty_tier)
print("\nExisting rows (loyalty_tier should be NULL):")
spark.sql("""
SELECT customer_id, loyalty_tier
FROM workspace.default.silver_customers
WHERE loyalty_tier IS NULL
LIMIT 5
""").show()
# Check new rows (should have 'gold' for loyalty_tier)
print("\nNew rows (loyalty_tier should be 'gold'):")
spark.sql("""
SELECT customer_id, loyalty_tier
FROM workspace.default.silver_customers
WHERE loyalty_tier IS NOT NULL
LIMIT 5
""").show()What should I see?
The DESCRIBE TABLE output now includes loyalty_tier as a string column. The first query shows existing rows with NULL in the new column. The second query shows the 10 newly appended rows with gold populated.
This is the key behavior: schema evolution adds columns non-destructively. Existing data is never modified. Only new writes populate the new column.
- Run the cell. Confirm you see loyalty_tier in the schema output and both NULL and gold values in the respective queries.
Not seeing the loyalty_tier column?
Make sure you ran the cell with .option("mergeSchema", "true") successfully before this verification cell. Check the cell output above for any errors.
help me troubleshoot why my schema evolution write didn't add the loyalty_tier column
Simulate a complex upstream schema change
In reality, upstream systems rarely add just one column at a time. A new feature release might add several fields simultaneously. Delta Lake handles this just as gracefully.
- Add a new cell to simulate multiple new columns arriving from an upstream system at once:
from pyspark.sql.functions import lit, current_date, rand
# Simulate an upstream system sending customer records with 3 new fields
# These represent a CRM upgrade that now tracks segmentation data
upstream_batch = (
silver_customers
.limit(15)
.withColumn("customer_segment", lit("premium"))
.withColumn("last_purchase_date", current_date())
.withColumn("lifetime_value", (rand() * 1000).cast("double"))
)
print(f"Upstream batch columns: {upstream_batch.columns}")
print(f"New columns being added: customer_segment, last_purchase_date, lifetime_value")What does this code do?
You're creating a batch that simulates what a CRM system upgrade would send. Three new columns arrive at once: customer_segment (a text label), last_purchase_date (a date), and lifetime_value (a numeric value). This is a common production scenario where an upstream team ships new features without coordinating a schema migration.
- Run the cell to confirm the batch has the three new columns listed.
- Add a new cell to write this batch with schema evolution and verify the result:
# Append the batch with multiple new columns using schema evolution
upstream_batch.write.mode("append").option("mergeSchema", "true").saveAsTable("workspace.default.silver_customers")
print("Complex schema evolution succeeded!")
print("\nUpdated table schema:")
spark.sql("DESCRIBE TABLE workspace.default.silver_customers").show(truncate=False)
# Show a sample of data with all new columns
print("\nSample showing all evolved columns:")
spark.sql("""
SELECT customer_id, loyalty_tier, customer_segment, last_purchase_date, lifetime_value
FROM workspace.default.silver_customers
WHERE customer_segment IS NOT NULL
LIMIT 5
""").show(truncate=False)What should I see?
The DESCRIBE TABLE output now shows all four new columns: loyalty_tier, customer_segment, last_purchase_date, and lifetime_value. The sample query shows the 15 newly appended rows with values in all new columns.
Delta Lake added three columns in a single write operation without breaking any existing data or requiring a table rebuild. Every existing row still has NULL for the new columns.
- Run the cell. Confirm the schema now includes all four new columns and the sample data shows populated values for the latest rows.
- Add a final markdown cell comparing this experience to plain Parquet:
# %md
# ## Schema Evolution vs Plain Parquet
#
# | Scenario | Delta Lake | Plain Parquet |
# |----------|-----------|---------------|
# | New column arrives | `.option("mergeSchema", "true")` adds it automatically | Must rewrite ALL existing files with new schema |
# | Existing data | Stays untouched, gets NULL for new columns | Must be reprocessed to include new column |
# | Multiple new columns at once | Handled in one write operation | Requires manual schema definition + full rewrite |
# | Accidental schema change | Blocked by default (schema enforcement) | Silently creates inconsistent file schemas |
# | Schema history | Tracked in _delta_log, visible in DESCRIBE HISTORY | No history, no rollback possible |
#
# **Key takeaway**: Delta Lake gives you a safety net (enforcement) AND a migration path (evolution).
# Plain Parquet gives you neither. You'd have to build custom tooling to detect schema drift,
# rewrite all existing files, and coordinate downstream consumers. Delta handles it in one line.Why does this matter in production?
With plain Parquet files, adding a column means rewriting every existing file with the new schema. If you have millions of records, that's an expensive full-table rewrite just because someone added one field.
Delta Lake handles it as a metadata operation. The transaction log records the schema change, existing files remain untouched, and all readers automatically see NULL for the new column on old rows. No rewrite needed.
- Run the markdown cell so it renders as formatted documentation in your notebook.
Seeing a quota limit error?
Databricks Free Edition has daily compute quotas. If you hit the limit, your compute shuts down for the rest of the day but your data and notebooks are preserved. Come back tomorrow to continue.
help me understand Databricks Free Edition quota limits and what to do when I hit them
Your Silver customers table now has a fully evolved schema with four new columns added non-destructively. Next up, you'll build Gold layer analytics that put all this cleaned, enriched data to work for business insights.
Build the Gold Layer
Your Silver layer now holds clean, enriched, and deduplicated data. The final layer of the Medallion Architecture transforms this data into business-ready analytics tables.
The Gold layer answers specific business questions. Instead of querying raw joins every time, you pre-compute aggregate tables that are fast, focused, and ready for dashboards or stakeholders.
In this step, get ready to:
- Build three Gold aggregate tables from your enriched Silver data.
- Run analytical SQL queries with built-in Databricks visualizations.
- Optimize Gold tables for production read performance.
Create the Gold aggregate tables
Gold tables are pre-aggregated for specific business use cases. You will build three tables: monthly revenue by product category, customer RFM segmentation, and seller performance metrics.
- In your Databricks workspace, click New in the sidebar, then select Notebook.
- Name the notebook 06_gold_layer and make sure the language is set to Python.
Start by building the monthly revenue table. This aggregates payment values by product category and month.
- Add this code to the first cell of your notebook and run it:
from pyspark.sql.functions import col, sum, count, avg, datediff, max, year, month, round
from delta.tables import DeltaTable
# Read the enriched Silver table
orders_enriched = spark.read.table("workspace.default.silver_orders_enriched")
# Build monthly revenue by product category
monthly_revenue = (
orders_enriched
.withColumn("order_year", year(col("order_purchase_timestamp")))
.withColumn("order_month", month(col("order_purchase_timestamp")))
.groupBy("order_year", "order_month", "product_category_name")
.agg(
round(sum("payment_value"), 2).alias("total_revenue"),
count("order_id").alias("order_count")
)
.orderBy("order_year", "order_month")
)
# Write as a Gold Delta table
monthly_revenue.write.mode("overwrite").saveAsTable("workspace.default.gold_monthly_revenue")
print("Gold monthly revenue table created.")
display(monthly_revenue.limit(10))What does this code do?
This cell reads from your enriched Silver orders table and groups the data by year, month, and product category. It calculates total revenue and order count for each group.
The result is written as a managed Delta Lake table called gold_monthly_revenue. Analysts can query this table directly without re-joining multiple tables every time.
- Run the cell. You should see a table displaying 10 rows with columns for year, month, product category, total revenue, and order count.
Next, build the customer RFM table. RFM stands for Recency, Frequency, and Monetary. It is a standard segmentation method in marketing analytics.
- Add a new cell with this code and run it:
from pyspark.sql.functions import current_date, countDistinct
# Read enriched orders for RFM calculation
orders_enriched = spark.read.table("workspace.default.silver_orders_enriched")
# Calculate RFM metrics per customer
customer_rfm = (
orders_enriched
.groupBy("customer_unique_id")
.agg(
# Recency: days since last purchase
datediff(current_date(), max("order_purchase_timestamp")).alias("recency_days"),
# Frequency: number of distinct orders
countDistinct("order_id").alias("frequency"),
# Monetary: total spend
round(sum("payment_value"), 2).alias("monetary_value")
)
)
# Write as a Gold Delta table
customer_rfm.write.mode("overwrite").saveAsTable("workspace.default.gold_customer_rfm")
print("Gold customer RFM table created.")
display(customer_rfm.limit(10))What is RFM segmentation?
RFM groups customers by three dimensions. Recency measures how many days since their last order. Frequency counts how many orders they placed. Monetary sums up their total spend.
Businesses use RFM to identify their best customers, those at risk of churning, and those who buy rarely but spend a lot. Your Gold table makes this analysis instant.
- Run the cell. You should see a table with customer IDs and their recency, frequency, and monetary values.
Finally, build the seller performance table. This measures how quickly sellers fulfill orders, how many they handle, and how much revenue they generate.
- Add a new cell with this code and run it:
# Build seller performance metrics
seller_performance = (
orders_enriched
.withColumn(
"delivery_days",
datediff(col("order_delivered_customer_date"), col("order_purchase_timestamp"))
)
.groupBy("seller_id")
.agg(
round(avg("delivery_days"), 1).alias("avg_delivery_days"),
countDistinct("order_id").alias("order_count"),
round(sum("payment_value"), 2).alias("total_revenue")
)
)
# Write as a Gold Delta table
seller_performance.write.mode("overwrite").saveAsTable("workspace.default.gold_seller_performance")
print("Gold seller performance table created.")
display(seller_performance.limit(10))What does this code do?
This cell calculates the number of days between purchase and delivery for each order. It then groups by seller and computes average delivery time, total order count, and total revenue per seller.
The gold_seller_performance table answers questions like "which sellers are fastest?" and "who generates the most revenue?" without querying the full enriched table.
- Run the cell. You should see a table with seller IDs, average delivery days, order counts, and revenue totals.
Seeing null values in delivery_days?
Some orders may not have been delivered yet, which means order_delivered_customer_date is null. This is normal for the Olist dataset. The avg function ignores nulls, so your average is still accurate.
If you see an error about column names not found, double-check that your silver_orders_enriched table contains the expected columns from the Silver layer step.
help me debug my Gold layer table creation
Run analytical queries with visualizations
Gold tables exist to be queried. You will now run analytical SQL queries against them and use Databricks built-in visualizations to create simple charts.
- Add a new cell with this code to find the top 10 product categories by revenue, then run it:
# Top 10 product categories by total revenue
top_categories = spark.sql("""
SELECT product_category_name,
SUM(total_revenue) as total_revenue,
SUM(order_count) as total_orders
FROM workspace.default.gold_monthly_revenue
GROUP BY product_category_name
ORDER BY total_revenue DESC
LIMIT 10
""")
display(top_categories)What does this query show?
This query aggregates across all months to find which product categories earned the most revenue overall. The display() function renders the results as a table, and you can switch to bar charts or other visualizations using the chart icon below the output.
- Run the cell. You should see a table with 10 product category names, their total revenue, and total orders.
- Click the chart icon (bar chart symbol) below the output table to switch to a visualization.
- Select a Bar chart with product_category_name on the x-axis and total_revenue on the y-axis.
- Add a new cell to query customer segments by state, then run it:
# Customer segments by state using RFM data joined with customer info
customer_segments = spark.sql("""
SELECT c.customer_state,
COUNT(*) as customer_count,
ROUND(AVG(r.monetary_value), 2) as avg_spend,
ROUND(AVG(r.frequency), 1) as avg_orders
FROM workspace.default.gold_customer_rfm r
JOIN workspace.default.silver_customers c
ON r.customer_unique_id = c.customer_unique_id
GROUP BY c.customer_state
ORDER BY customer_count DESC
""")
display(customer_segments)- Run the cell. You should see states ranked by customer count, with average spend and order frequency per state.
- Add a new cell for seller performance rankings, then run it:
# Seller performance rankings: top 10 sellers by revenue
seller_rankings = spark.sql("""
SELECT seller_id,
avg_delivery_days,
order_count,
total_revenue
FROM workspace.default.gold_seller_performance
ORDER BY total_revenue DESC
LIMIT 10
""")
display(seller_rankings)- Run the cell. You should see the top 10 sellers ranked by revenue, showing their delivery speed and order volume alongside.
Explore Databricks visualizations
Every display() output can be toggled between table and chart views. Try switching the seller rankings to a bar chart, or the customer segments to a different chart type using the visualization controls below each output.
Optimize Gold tables for production performance
When Delta tables are built from many small writes, they can accumulate small files that slow down reads. The OPTIMIZE command compacts these small files into larger ones for better query performance.
- Check the current file count by adding this cell and running it:
# Check file count BEFORE optimization
detail_before = spark.sql("DESCRIBE DETAIL workspace.default.gold_monthly_revenue")
print("Files BEFORE OPTIMIZE:")
display(detail_before.select("numFiles", "sizeInBytes"))What is DESCRIBE DETAIL?
This SQL command returns metadata about a Delta table in a single row. The numFiles column shows how many data files make up the current version of the table. The sizeInBytes column shows the total storage size.
- Run the cell. Note the numFiles value displayed. This is your "before" count.
- Run OPTIMIZE on all three Gold tables by adding this cell and running it:
# Run OPTIMIZE on all Gold tables to compact small files
gold_monthly = DeltaTable.forName(spark, "workspace.default.gold_monthly_revenue")
gold_monthly.optimize().executeCompaction()
gold_rfm = DeltaTable.forName(spark, "workspace.default.gold_customer_rfm")
gold_rfm.optimize().executeCompaction()
gold_seller = DeltaTable.forName(spark, "workspace.default.gold_seller_performance")
gold_seller.optimize().executeCompaction()
print("All Gold tables optimized.")What does OPTIMIZE do?
OPTIMIZE rewrites data files to produce evenly-balanced, larger files. This reduces the number of files a query needs to open, improving read performance. The operation is idempotent, meaning running it twice on the same data has no additional effect.
OPTIMIZE makes no changes to the data itself. Reads before and after OPTIMIZE return the same results. It only changes the physical layout of the files on storage.
- Run the cell. You should see the message "All Gold tables optimized." printed.
- Check the file count after optimization by adding this cell and running it:
# Check file count AFTER optimization
detail_after = spark.sql("DESCRIBE DETAIL workspace.default.gold_monthly_revenue")
print("Files AFTER OPTIMIZE:")
display(detail_after.select("numFiles", "sizeInBytes"))- Run the cell. Compare the numFiles value with what you recorded before. After optimization, you should see fewer (or equal) files.
What is liquid clustering?
Liquid clustering is a data layout optimization that replaces traditional partitioning and ZORDER. It automatically organizes data based on clustering keys to improve query performance. Databricks recommends it for all new tables.
You enable it with a CLUSTER BY clause when creating a table, or by altering an existing table with ALTER TABLE table_name CLUSTER BY (column_name). After enabling, run OPTIMIZE to apply the clustering.
For production workloads with frequent filtered queries (e.g., always filtering by date or category), liquid clustering groups related rows into the same files. This means Delta Lake can skip entire files that do not match your filter, dramatically reducing scan time.
- Add a markdown cell to your notebook documenting liquid clustering. Change the cell language to Markdown using the language dropdown at the top of the cell, then type the following:
## Liquid Clustering (Production Recommendation)
For production Gold tables with known query patterns, use liquid clustering:
```sql
-- Enable clustering on an existing table
ALTER TABLE workspace.default.gold_monthly_revenue CLUSTER BY (product_category_name)
-- Then run OPTIMIZE to apply clustering
OPTIMIZE workspace.default.gold_monthly_revenue
```
Liquid clustering replaces traditional partitioning and ZORDER.
Databricks recommends it for all Unity Catalog managed tables.
You can also use CLUSTER BY AUTO to let Databricks choose the best columns automatically.Why not apply liquid clustering here?
With the Olist dataset size (~100K orders), the Gold tables are small enough that OPTIMIZE alone provides good performance. Liquid clustering shines at larger scale where query patterns consistently filter on specific columns.
The documentation cell above records the pattern for reference when you work with larger datasets in production.
OPTIMIZE seems to have no effect?
If your before and after file counts are the same, your table was likely written as a single file already. This is normal for smaller datasets. OPTIMIZE is idempotent and has no effect if files are already well-sized.
If you see a quota-related error, your Databricks Free Edition daily compute quota may have been reached. Your data is preserved. Try again when the quota resets.
help me understand why OPTIMIZE did not reduce my file count
✔️ Awesome, I've got everything!
Great. Double check you have saved your notebook. All three Gold tables should be visible in Catalog Explorer under workspace.default.
ⓧ I'd like to double check the full code
Here is the complete code for the 06_gold_layer notebook:
from pyspark.sql.functions import col, sum, count, avg, datediff, max, year, month, round
from delta.tables import DeltaTable
# Read the enriched Silver table
orders_enriched = spark.read.table("workspace.default.silver_orders_enriched")
# Build monthly revenue by product category
monthly_revenue = (
orders_enriched
.withColumn("order_year", year(col("order_purchase_timestamp")))
.withColumn("order_month", month(col("order_purchase_timestamp")))
.groupBy("order_year", "order_month", "product_category_name")
.agg(
round(sum("payment_value"), 2).alias("total_revenue"),
count("order_id").alias("order_count")
)
.orderBy("order_year", "order_month")
)
# Write as a Gold Delta table
monthly_revenue.write.mode("overwrite").saveAsTable("workspace.default.gold_monthly_revenue")
print("Gold monthly revenue table created.")
display(monthly_revenue.limit(10))from pyspark.sql.functions import current_date, countDistinct
# Read enriched orders for RFM calculation
orders_enriched = spark.read.table("workspace.default.silver_orders_enriched")
# Calculate RFM metrics per customer
customer_rfm = (
orders_enriched
.groupBy("customer_unique_id")
.agg(
# Recency: days since last purchase
datediff(current_date(), max("order_purchase_timestamp")).alias("recency_days"),
# Frequency: number of distinct orders
countDistinct("order_id").alias("frequency"),
# Monetary: total spend
round(sum("payment_value"), 2).alias("monetary_value")
)
)
# Write as a Gold Delta table
customer_rfm.write.mode("overwrite").saveAsTable("workspace.default.gold_customer_rfm")
print("Gold customer RFM table created.")
display(customer_rfm.limit(10))# Build seller performance metrics
seller_performance = (
orders_enriched
.withColumn(
"delivery_days",
datediff(col("order_delivered_customer_date"), col("order_purchase_timestamp"))
)
.groupBy("seller_id")
.agg(
round(avg("delivery_days"), 1).alias("avg_delivery_days"),
countDistinct("order_id").alias("order_count"),
round(sum("payment_value"), 2).alias("total_revenue")
)
)
# Write as a Gold Delta table
seller_performance.write.mode("overwrite").saveAsTable("workspace.default.gold_seller_performance")
print("Gold seller performance table created.")
display(seller_performance.limit(10))# Top 10 product categories by total revenue
top_categories = spark.sql("""
SELECT product_category_name,
SUM(total_revenue) as total_revenue,
SUM(order_count) as total_orders
FROM workspace.default.gold_monthly_revenue
GROUP BY product_category_name
ORDER BY total_revenue DESC
LIMIT 10
""")
display(top_categories)# Customer segments by state using RFM data joined with customer info
customer_segments = spark.sql("""
SELECT c.customer_state,
COUNT(*) as customer_count,
ROUND(AVG(r.monetary_value), 2) as avg_spend,
ROUND(AVG(r.frequency), 1) as avg_orders
FROM workspace.default.gold_customer_rfm r
JOIN workspace.default.silver_customers c
ON r.customer_unique_id = c.customer_unique_id
GROUP BY c.customer_state
ORDER BY customer_count DESC
""")
display(customer_segments)# Seller performance rankings: top 10 sellers by revenue
seller_rankings = spark.sql("""
SELECT seller_id,
avg_delivery_days,
order_count,
total_revenue
FROM workspace.default.gold_seller_performance
ORDER BY total_revenue DESC
LIMIT 10
""")
display(seller_rankings)# Check file count BEFORE optimization
detail_before = spark.sql("DESCRIBE DETAIL workspace.default.gold_monthly_revenue")
print("Files BEFORE OPTIMIZE:")
display(detail_before.select("numFiles", "sizeInBytes"))# Run OPTIMIZE on all Gold tables to compact small files
gold_monthly = DeltaTable.forName(spark, "workspace.default.gold_monthly_revenue")
gold_monthly.optimize().executeCompaction()
gold_rfm = DeltaTable.forName(spark, "workspace.default.gold_customer_rfm")
gold_rfm.optimize().executeCompaction()
gold_seller = DeltaTable.forName(spark, "workspace.default.gold_seller_performance")
gold_seller.optimize().executeCompaction()
print("All Gold tables optimized.")# Check file count AFTER optimization
detail_after = spark.sql("DESCRIBE DETAIL workspace.default.gold_monthly_revenue")
print("Files AFTER OPTIMIZE:")
display(detail_after.select("numFiles", "sizeInBytes"))The final cell is a Markdown cell documenting liquid clustering. Switch the cell language to Markdown and add the documentation about CLUSTER BY and CLUSTER BY AUTO.
Your Gold layer is complete. You now have a full Medallion Architecture pipeline from raw CSV files through Bronze and Silver to business-ready Gold tables. Next up, you will export everything to GitHub and document your work as a portfolio project.
Export Notebooks and Publish to GitHub
You have a complete Medallion Architecture lakehouse running in Databricks. Your notebooks demonstrate every core Delta Lake feature from ACID transactions to schema evolution.
To make this work shareable and portfolio-ready, you need to get it out of Databricks and into a GitHub repository. In this step, you will export your notebooks as plain Python files, write a professional README, and push everything to GitHub.
In this step, get ready to:
- Export all 6 notebooks from Databricks as .py source files.
- Write a comprehensive README.md documenting your project.
- Push everything to a GitHub repository.
Export all notebooks as Python source files
Databricks notebooks can be exported as .py source files. This format preserves your code and cell structure as plain Python that anyone can read on GitHub without needing Databricks.
- In your Databricks workspace, open the 01_delta_vs_parquet notebook.
- Click File in the notebook toolbar.
- Click Export... from the dropdown.
- Select Source file from the export file type dropdown.
- Click Export. The file downloads to your computer as 01_delta_vs_parquet.py.
- Repeat this process for each of the remaining 5 notebooks:
- 02_bronze_layer
- 03_silver_layer
- 04_time_travel
- 05_schema_evolution
- 06_gold_layer
What is a Source file export?
When you export as Source file, Databricks produces a .py file that contains only your code statements. The first line will be # Databricks notebook source and cell boundaries are marked with # COMMAND ---------- comments.
This makes the file readable as plain Python on GitHub while still being re-importable into Databricks as a notebook.
- Once all 6 files are downloaded, open your local file system and move them into a single folder. You will organize them into the final project structure next.
Now create the local project folder structure to organize these exports.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and the notebooks subdirectory by running these commands:
mkdir delta-lake-lakehouse
cd delta-lake-lakehouse
mkdir notebooks- Move all 6 downloaded .py files into the notebooks/ directory.
- Confirm the files are in place by running:
ls notebooks/You should see all 6 files listed: 01_delta_vs_parquet.py, 02_bronze_layer.py, 03_silver_layer.py, 04_time_travel.py, 05_schema_evolution.py, 06_gold_layer.py.
Don't see all 6 files?
Make sure you exported each notebook individually using File > Export... > Source file. Check your browser's Downloads folder for the files and move them into the notebooks/ directory.
Help me find my exported notebook files.
Write a comprehensive README.md
A good README is what transforms a collection of code files into a portfolio project. It tells readers what you built, why it matters, and how to reproduce it.
- Create a file called README.md in the root of your delta-lake-lakehouse folder (not inside notebooks/). Open it in any text editor.
- Write the following sections in your README. Use the template below as a starting point and customize it with details from your own notebooks:
# Delta Lake Lakehouse: Brazilian E-Commerce Data Pipeline
## Project Overview
A complete data lakehouse built on Databricks Free Edition, processing the Brazilian E-Commerce (Olist) dataset through Bronze, Silver, and Gold layers using Delta Lake. Demonstrates ACID transactions, time travel, schema evolution, MERGE operations, and OPTIMIZE.
## Architecture
```
CSV Files (Kaggle)
|
v
+-------------+ +-------------+ +-------------+
| BRONZE | --> | SILVER | --> | GOLD |
| Raw Ingestion| | Clean + Join| | Aggregates |
| + Metadata | | + MERGE | | + OPTIMIZE |
+-------------+ +-------------+ +-------------+
| | |
v v v
ACID writes Time Travel Business KPIs
Audit trail Schema Evol. RFM Segments
```
## Dataset
**Brazilian E-Commerce Public Dataset by Olist** (Kaggle)
- ~100K orders from 2016-2018
- 8 interconnected CSV files: orders, order_items, customers, products, sellers, payments, reviews, category translations
- Source: kaggle.com/datasets/olistbr/brazilian-ecommerce
## Setup Instructions
1. Sign up for [Databricks Free Edition](https://www.databricks.com/try-databricks)
2. Download the Olist dataset from Kaggle
3. Create a Unity Catalog volume named `olist-data` in `workspace.default`
4. Upload all 8 CSV files to the volume
5. Import the notebooks from the `notebooks/` directory into your workspace
## Notebooks
| # | Notebook | Description |
|---|----------|-------------|
| 1 | `01_delta_vs_parquet.py` | Hands-on comparison showing why Delta Lake's transaction log changes everything |
| 2 | `02_bronze_layer.py` | Raw ingestion of all 8 tables with ACID guarantees and metadata columns |
| 3 | `03_silver_layer.py` | Data cleaning, deduplication, MERGE/upsert operations, and enriched joins |
| 4 | `04_time_travel.py` | Query historical versions, simulate data disaster, RESTORE, and VACUUM |
| 5 | `05_schema_evolution.py` | Schema enforcement vs. evolution with mergeSchema |
| 6 | `06_gold_layer.py` | Business analytics aggregates with OPTIMIZE for performance |
## Key Learnings
- **Delta Lake = Parquet + Transaction Log**: The `_delta_log` directory enables ACID, time travel, and DML operations
- **ACID Transactions**: Partial writes never corrupt tables; failed writes leave data at last valid version
- **Time Travel**: Query any historical version and RESTORE after accidental deletes
- **Schema Evolution**: Add new columns to existing tables without breaking queries using `mergeSchema`
- **MERGE (Upsert)**: Atomic insert/update/delete in a single operation
- **OPTIMIZE**: Compact small files for better read performance in production
## How to Reproduce
1. Follow the Setup Instructions above
2. Run notebooks in order (01 through 06)
3. Each notebook is self-contained with inline documentation
4. Requires Databricks Free Edition (serverless compute, Python and SQL only)
What makes a good data engineering README?
The architecture diagram (even as ASCII text) immediately shows readers you think in systems. The notebook table gives quick navigation. The Key Learnings section proves you understand the concepts, not just the commands.
Customize the template with specifics from your own implementation. Mention exact record counts, any interesting findings from your Gold layer queries, or challenges you solved.
- Save README.md.
- Confirm your folder structure looks correct by running:
lsYou should see README.md and notebooks listed at the root of your delta-lake-lakehouse folder.
Initialize Git and push to GitHub
With your files organized, you can now version-control the project and publish it to GitHub where it becomes part of your public portfolio.
- Make sure you are inside the delta-lake-lakehouse folder in your terminal.
- Initialize a Git repository by running:
git init- Stage all files by running:
git add .- Commit the files by running:
git commit -m "Initial commit: Delta Lake lakehouse project with 6 notebooks"What do these Git commands do?
- git init creates a new Git repository in your current directory.
- git add . stages every file in the folder for the next commit.
- git commit -m "..." saves a snapshot of all staged files with a descriptive message.
- Go to github.com in your browser.
- Click the + icon in the top-right corner.
- Select New repository.
- Enter delta-lake-lakehouse as the repository name.
- Leave the repository set to Public.
- Do NOT check any of the initialization options (no README, no .gitignore, no license). You already have a README locally.
- Click Create repository.
GitHub shows you setup commands for pushing an existing repository. Use those commands in your terminal.
- Connect your local repository to GitHub and push by running these commands (replace the URL with your own repository URL from GitHub):
git branch -M main
git remote add origin [[REPO_URL="https://github.com/your-username/delta-lake-lakehouse.git"]]
git push -u origin mainWhat do these commands do?
- git branch -M main renames your default branch to main.
- git remote add origin connects your local repo to the GitHub repository.
- git push -u origin main uploads all your committed files to GitHub.
- Refresh your GitHub repository page in the browser.
- Verify that the README renders properly with the architecture diagram and notebook table visible.
- Click into the notebooks/ folder and open any .py file to confirm it is readable as plain Python with your code and markdown comments intact.
Push failing with authentication error?
GitHub no longer accepts password authentication for Git operations. You need either a Personal Access Token (PAT) or SSH key configured. When prompted for a password, paste your PAT instead.
Help me set up GitHub authentication for pushing.
Your Delta Lake lakehouse project is now live on GitHub. Anyone can read your notebooks, understand your architecture from the README, and reproduce the entire pipeline on their own Databricks workspace.
Secret mission
Build a Change Data Feed (CDF) Audit Pipeline
Your lakehouse tracks table versions through Delta's transaction log. But what if downstream systems need to know exactly which rows changed between versions? Build a Change Data Feed pipeline that captures every INSERT, UPDATE, and DELETE at the row level, then aggregate those changes into a Gold-layer audit 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 entirely on Databricks Free Edition with no external cloud resources, so there are no ongoing costs.
Resources you used:
- Databricks Free Edition workspace (serverless compute, no cost).
- Unity Catalog managed Delta tables in the workspace.default schema (22 tables across Bronze, Silver, and Gold layers).
- Unity Catalog managed volume olist-data containing 8 CSV files.
- 7 Databricks notebooks (01 through 07).
- Local project folder with exported .py files and README.md.
- GitHub repository with notebooks and documentation.
Keep everything running
No action needed. Choose this if you plan to continue experimenting with your Delta Lake tables or want to reference your notebooks later.
- Your Free Edition workspace persists indefinitely at no cost. Databricks may delete accounts that are inactive for a prolonged period.
- Serverless compute stops automatically when not in use. You are never charged.
- Your Delta tables, volume, and notebooks remain accessible whenever you log back in.
- Your local project folder and GitHub repository stay as they are.
Pause - I'll come back to this later
No action needed. Serverless compute stops automatically when not in use, so there is nothing to shut down or pause manually.
- Your workspace, tables, and notebooks persist until you return.
- Daily compute quotas reset each day. If you hit a quota limit during your session, compute becomes available again the next day.
- Your local files and GitHub repository remain unchanged.
Delete - I don't want to use this again
Remove all project resources and start fresh. Follow the steps below to clean up everything created during this project.
Drop all Delta tables
- Open a new notebook in your Databricks workspace and run the following SQL to drop all tables you created:
DROP TABLE IF EXISTS workspace.default.bronze_orders;
DROP TABLE IF EXISTS workspace.default.bronze_customers;
DROP TABLE IF EXISTS workspace.default.bronze_order_items;
DROP TABLE IF EXISTS workspace.default.bronze_products;
DROP TABLE IF EXISTS workspace.default.bronze_order_payments;
DROP TABLE IF EXISTS workspace.default.bronze_order_reviews;
DROP TABLE IF EXISTS workspace.default.bronze_sellers;
DROP TABLE IF EXISTS workspace.default.bronze_product_category_name_translation;
DROP TABLE IF EXISTS workspace.default.silver_orders;
DROP TABLE IF EXISTS workspace.default.silver_customers;
DROP TABLE IF EXISTS workspace.default.silver_order_items;
DROP TABLE IF EXISTS workspace.default.silver_products;
DROP TABLE IF EXISTS workspace.default.silver_order_payments;
DROP TABLE IF EXISTS workspace.default.silver_order_reviews;
DROP TABLE IF EXISTS workspace.default.silver_sellers;
DROP TABLE IF EXISTS workspace.default.silver_orders_enriched;
DROP TABLE IF EXISTS workspace.default.gold_monthly_revenue;
DROP TABLE IF EXISTS workspace.default.gold_customer_rfm;
DROP TABLE IF EXISTS workspace.default.gold_seller_performance;
DROP TABLE IF EXISTS workspace.default.gold_customer_audit;Delete the Unity Catalog volume
- Click Catalog in the Databricks sidebar.
- Expand workspace > default > Volumes.
- Click the kebab menu next to olist-data and select Delete.
- Confirm the deletion in the dialog.
Delete Databricks notebooks
- In the Databricks sidebar, click Workspace.
- For each of the 7 notebooks (01_delta_vs_parquet through 07_change_data_feed), right-click the notebook and select Move to Trash.
Delete local project files
- Delete your local project folder by running this command in your terminal:
rm -rf ~/Desktop/delta-lake-projectOn Windows, use:
Remove-Item -Recurse -Force ~\Desktop\delta-lake-projectDelete the GitHub repository
- Navigate to your repository on GitHub.
- Click Settings in the repository navigation bar.
- Scroll down to the Danger Zone section and click Delete this repository.
- Follow the confirmation prompts: click I want to delete this repository, then I have read and understand these effects, type the repository name, and click Delete this repository.
That's a Wrap!
That's a Wrap!
Nice work! You just built a complete Delta Lake lakehouse from scratch, processing real Brazilian e-commerce data through a full Medallion Architecture on Databricks Free Edition.
You've learned how to:
- Build a Medallion Architecture (Bronze, Silver, Gold) with Delta Lake, ingesting real e-commerce data with ACID guarantees and transforming it through MERGE/upsert operations.
- Recover from data disasters using time travel and RESTORE, and handle real-world data drift with schema evolution.
- Export a complete GitHub-ready portfolio project with organized notebooks, Gold layer analytics using OPTIMIZE, and a comprehensive README documenting your architecture.
- Built a Change Data Feed audit pipeline that tracks row-level changes for incremental processing.
Ready to quiz yourself?