Healthcare Data Lakehouse with Delta Lake

Build a medallion architecture lakehouse on synthetic healthcare data locally.

Introduction

30 Second Summary

Every hospital visit, prescription, and diagnosis generates data that needs to be stored, cleaned, and made ready for analysis. Data engineers build the pipelines that transform this raw information into reliable, query-ready tables.

In this project, you will build a local data lakehouse that processes synthetic healthcare patient data through three layers (Bronze, Silver, and Gold) using Delta Lake and PySpark. The entire environment runs inside Docker, making it portable and reproducible.

What You'll Build

A Dockerized healthcare data lakehouse implementing the medallion architecture pattern with Delta Lake, PySpark, and Jupyter Notebook.

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

  • A Docker environment running PySpark, Delta Lake, and Jupyter Notebook for portable data engineering.
  • A medallion architecture pipeline that ingests raw CSV data into Bronze, cleans it into Silver, and aggregates it into Gold Delta tables.
  • Working demonstrations of schema evolution, MERGE upserts, and time travel queries on Delta tables.
  • Secret Mission: Add date-based partitioning to your Silver layer and measure the query performance improvement.

Are there any prerequisites?

This project assumes you have Docker Desktop installed and running, Python installed, and VS Code installed. Step 1 walks you through creating the project files and building the Docker environment from scratch.

Set Up the Docker Environment

With your project files ready, it's time to bring the environment to life. Your Dockerfile and docker-compose.yml define a container that packages PySpark, Delta Lake, and Jupyter Notebook together.

In this step, you will build and start that container, verify everything works, and generate the synthetic healthcare data your lakehouse will process.

In this step, get ready to:
  • Build and launch the Docker container with PySpark and Delta Lake.
  • Verify that Spark and Delta Lake work correctly inside the container.
  • Generate synthetic healthcare data for the lakehouse pipeline.
Review your Docker configuration

Your healthcare-lakehouse folder already contains the infrastructure files that define your environment. Before building anything, take a moment to understand what each file does.

What do these files do?

  • Dockerfile starts from the jupyter/pyspark-notebook:spark-3.5.0 base image and installs delta-spark==3.2.0 via requirements.txt.
  • docker-compose.yml maps port 8888 for Jupyter and port 4040 for the Spark UI. It mounts your local notebooks/, data/, and scripts/ folders into the container.
  • scripts/generate_data.py produces synthetic patient, encounter, and condition CSV files modeled on the Synthea healthcare data format.
Build and launch the Docker environment

Docker will download the base image, install your Python dependencies, and start the Jupyter server. The first build takes a few minutes because it pulls the PySpark image (several GB).

  • In your terminal, make sure you are inside the healthcare-lakehouse folder by running:
cd ~/Desktop/healthcare-lakehouse
  • Build the Docker image and start the container in detached mode by running:
docker-compose up --build -d

What does this command do?

  • --build tells Docker to build the image from your Dockerfile before starting the container.
  • -d runs the container in the background (detached mode) so your terminal stays free.
  • Confirm the container is running by running:
docker ps

You should see a container named healthcare-lakehouse with status Up in the output.

Don't see the container running?

  • Make sure Docker Desktop is running. The Docker daemon must be active before you can build or start containers.
  • If you see a port conflict error mentioning 8888, another Jupyter instance may already be using that port. Stop it first or change the port mapping in docker-compose.yml.
  • Run docker-compose logs to inspect any build errors.

Help me debug my Docker container startup

  • Open your browser and navigate to http://localhost:8888.
  • When prompted for a token, enter lakehouse and click Log in.

You should see the Jupyter file browser showing your notebooks/, data/, and scripts/ folders.

Next, verify that PySpark and Delta Lake are configured correctly by running a quick test.

  • In Jupyter, click New in the top right, then select Python 3 (ipykernel) to create a new notebook.
  • Paste the following code into the first cell and run it with Shift+Enter:
# Test PySpark + Delta Lake configuration
from delta import *
from pyspark.sql import SparkSession

# Build SparkSession with Delta Lake extensions
builder = (
    SparkSession.builder
    .appName("DeltaTest")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog"
    )
    .master("local[*]")
)

spark = configure_spark_with_delta_pip(builder).getOrCreate()

# Write and read a small Delta table to confirm everything works
test_data = spark.createDataFrame([(1, "test"), (2, "delta")], ["id", "value"])
test_data.write.format("delta").mode("overwrite").save("/tmp/delta_test")
result = spark.read.format("delta").load("/tmp/delta_test")
result.show()
print(f"Delta Lake test passed! Read {result.count()} rows.")

# Stop the test session
spark.stop()

What does this test do?

  • configure_spark_with_delta_pip(builder) downloads the Delta Lake JARs from Maven and attaches them to your Spark session. This may take 1-2 minutes on first run.
  • The test writes a tiny DataFrame to Delta format, then reads it back. If both operations succeed, your environment is correctly wired.
  • spark.sql.extensions and spark.sql.catalog.spark_catalog are the two config keys Delta Lake requires to register its SQL features.

After the cell finishes, you should see a small table with two rows (id and value columns) and the message Delta Lake test passed! Read 2 rows..

Seeing a Java or JAR-related error?

  • The first time you run configure_spark_with_delta_pip, it downloads JARs from Maven Central. Make sure your container has internet access.
  • If you see a NoSuchMethodError, confirm your requirements.txt specifies delta-spark==3.2.0 (not 3.2.1, which has a known compatibility bug with Spark 3.5.x).
  • If the cell hangs, wait 2-3 minutes. The JAR download is slow on the first run.

Help me fix this Java or JAR error

Generate the synthetic healthcare dataset

Your lakehouse needs data to process. The generate_data.py script creates 1000 synthetic patient records along with their encounters and medical conditions.

  • Run the data generation script inside the container by running this command in your terminal:
docker exec healthcare-lakehouse python scripts/generate_data.py

What does this command do?

  • docker exec runs a command inside an already-running container.
  • healthcare-lakehouse is the container name defined in your docker-compose.yml.
  • The script generates three CSV files matching the Synthea healthcare data format: patients.csv, encounters.csv, and conditions.csv.

You should see output confirming how many records were written to each file.

  • Verify the CSV files were created by running:
ls data/raw/

You should see patients.csv, encounters.csv, and conditions.csv listed in the output.

Files not showing up?

  • Make sure you ran the docker exec command, not python scripts/generate_data.py directly. The script must run inside the container where the volume mount points to your local data/raw/ folder.
  • Check that your container is still running with docker ps. If it stopped, restart it with docker-compose up -d.

Help me troubleshoot the missing CSV files

What data did the script generate?

  • patients.csv: 1000 synthetic patient records with demographics, addresses, and financial data.
  • encounters.csv: multiple healthcare encounters per patient (office visits, emergency, inpatient stays) with timestamps and costs.
  • conditions.csv: medical conditions diagnosed during encounters (diabetes, hypertension, sinusitis, etc.).

Your environment is fully operational with synthetic healthcare data ready to process. Next up, you will ingest that raw CSV data into your Bronze layer.

Build the Bronze Layer

Your Docker environment is running and your synthetic healthcare data is ready.

Now it's time to build the first layer of your medallion architecture: the Bronze layer. The Bronze layer is the raw ingestion layer. Its job is to capture source data exactly as it arrives, stamp it with a load timestamp, and store it in Delta Lake format so downstream layers can rely on ACID guarantees.

In this step, get ready to:
  • Read raw CSV files into Spark DataFrames and inspect their schemas.
  • Write raw data to Delta format in the Bronze layer with ingestion timestamps.
  • Verify Bronze tables with Spark SQL queries.
Create the lakehouse notebook and initialize Spark

Every cell you write in this project will live in a single Jupyter notebook. This notebook becomes the spine of your lakehouse pipeline.

  • Switch back to the JupyterLab tab in your browser (running at http://localhost:8888).
  • In the Launcher, click Python 3 under the Notebook section to create a new notebook.
  • Right-click the Untitled.ipynb tab at the top and select Rename Notebook.
  • Name it healthcare_lakehouse and press Enter.

Your notebook starts with one empty code cell. You will add each chunk of code below into its own cell.

  • Initialize PySpark with Delta Lake support by pasting this code into the first cell:
# Initialize Spark with Delta Lake extensions
import pyspark
from delta import *
from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, DateType, TimestampType
)
from pyspark.sql.functions import (
    col, year, month, to_date, to_timestamp, when, lit,
    count, avg, round as spark_round, current_timestamp,
    datediff, expr, desc
)

# Build the SparkSession with Delta Lake catalog and extensions
builder = (
    SparkSession.builder
    .appName("HealthcareLakehouse")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog"
    )
    .master("local[*]")
)

# configure_spark_with_delta_pip downloads Delta JARs and creates the session
spark = configure_spark_with_delta_pip(builder).getOrCreate()
spark.sparkContext.setLogLevel("WARN")
print(f"Spark version: {spark.version}")
print("Delta Lake configured successfully!")

What does this code do?

This cell does three things to set up your lakehouse environment.

  • Imports all the PySpark and Delta Lake modules you will use across the entire project (functions, types, and the DeltaTable class for MERGE operations later).
  • Configures the SparkSession builder with Delta Lake's SQL extension and catalog so Spark understands the Delta table format natively.
  • Calls configure_spark_with_delta_pip which downloads the Delta Lake JARs from Maven and attaches them to your Spark runtime.
  • Run the cell by pressing Shift+Enter.

The first run may take 1-2 minutes while Delta Lake JARs download from Maven. Once complete, you should see output showing your Spark version and the message Delta Lake configured successfully!.

Cell stuck or throwing a Java error?

  • If the cell hangs for more than 3 minutes, your container may not have internet access. Check that Docker Desktop is connected to the network.
  • If you see a NoSuchMethodError, confirm your requirements.txt has delta-spark==3.2.0 (not 3.2.1) and rebuild your container.

Help me debug a Java error in my PySpark Delta Lake setup.

  • Define your data layer paths by pasting this code into a new cell:
# Define paths for each layer of the medallion architecture
BRONZE_PATH = "/home/jovyan/work/data/bronze"
SILVER_PATH = "/home/jovyan/work/data/silver"
GOLD_PATH = "/home/jovyan/work/data/gold"
RAW_PATH = "/home/jovyan/work/data/raw"

What are these paths?

Each constant maps to a directory inside your Docker container where that layer's Delta tables will live. The RAW_PATH points to the CSV files you generated in Step 1. The other three paths are where you will write Delta tables in Bronze, Silver, and Gold layers.

  • Run the cell with Shift+Enter. No output is expected. The cell should execute without errors and a number will appear next to it.
Read raw CSV files into Spark DataFrames

With Spark initialized, you can now read the three CSV files your data generation script created. Spark will infer the column types automatically from the CSV content.

  • Load the raw CSV files into DataFrames by pasting this code into a new cell:
# Read raw CSV files with headers and inferred schemas
patients_raw = spark.read.csv(
    f"{RAW_PATH}/patients.csv", header=True, inferSchema=True
)
encounters_raw = spark.read.csv(
    f"{RAW_PATH}/encounters.csv", header=True, inferSchema=True
)
conditions_raw = spark.read.csv(
    f"{RAW_PATH}/conditions.csv", header=True, inferSchema=True
)

# Print row counts to verify data loaded correctly
print(f"Patients: {patients_raw.count()} rows")
print(f"Encounters: {encounters_raw.count()} rows")
print(f"Conditions: {conditions_raw.count()} rows")
patients_raw.printSchema()

What does this code do?

  • spark.read.csv() reads each CSV file into a Spark DataFrame. The header=True flag tells Spark the first row contains column names, and inferSchema=True makes Spark scan the data to detect column types (string, integer, double).
  • .count() triggers Spark to read all rows and return the total. This confirms data loaded successfully.
  • .printSchema() displays the column names and inferred types so you can verify the structure matches what you expect.
  • Run the cell with Shift+Enter.

You should see Patients: 1000 rows printed first, followed by encounter and condition counts, then the schema tree showing columns like Id, BIRTHDATE, GENDER, etc.

Seeing 0 rows or a file not found error?

  • Check that you ran the data generation script in Step 1. The CSV files must exist in data/raw/ inside the container.
  • If you restarted the container since Step 1, the data/ volume mount preserves your files. Re-run the generate script if they are missing.

Help me debug why Spark can't find my CSV files.

Write raw data to Delta format in the Bronze layer

Reading CSVs into DataFrames is temporary. To make this data durable and queryable with ACID guarantees, you need to persist it as Delta tables. The Bronze layer stores data exactly as it arrived, plus a metadata column tracking when each batch was ingested.

  • Write the Bronze Delta tables by pasting this code into a new cell:
# Add ingestion timestamp to track when data was loaded
patients_bronze = patients_raw.withColumn("_ingested_at", current_timestamp())
encounters_bronze = encounters_raw.withColumn("_ingested_at", current_timestamp())
conditions_bronze = conditions_raw.withColumn("_ingested_at", current_timestamp())

# Write each DataFrame as a Delta table to the Bronze layer
patients_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/patients"
)
encounters_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/encounters"
)
conditions_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/conditions"
)
print("Bronze layer written successfully!")

What does this code do?

  • .withColumn("_ingested_at", current_timestamp()) adds a new column to each DataFrame containing the exact time the data was loaded. This is a common data engineering pattern for auditing and debugging pipeline runs.
  • .write.format("delta").mode("overwrite").save() persists the DataFrame as a Delta table on disk. The overwrite mode replaces any existing data at that path, making the operation idempotent (safe to re-run).

Why add _ingested_at?

In production pipelines, data arrives in batches. The _ingested_at timestamp lets you answer questions like "which records came in today's load?" or "when did this row first appear?" This is essential for debugging data quality issues and tracking pipeline freshness.

  • Run the cell with Shift+Enter.

You should see Bronze layer written successfully! printed after a few seconds. Spark writes Parquet files plus a _delta_log/ transaction log into each Bronze directory.

Seeing a permission denied or path error?

  • The data/bronze/ directories are created automatically by Spark on write. If you get a permission error, check that your Docker volume mount maps ./data correctly in docker-compose.yml.
  • If you see a "delta" format not recognized error, your SparkSession was not initialized with Delta extensions. Re-run Cell 1 to reinitialize.

Help me fix a write error when saving Delta tables.

Verify Bronze tables with Spark SQL

Writing data is only half the job. You need to confirm the Delta tables are readable and contain the expected data. Registering them as temporary views lets you query them with Spark SQL, just like querying a database.

  • Register the Bronze tables as SQL views and run verification queries by pasting this code into a new cell:
# Register Delta tables as temporary views for SQL queries
spark.read.format("delta").load(f"{BRONZE_PATH}/patients").createOrReplaceTempView(
    "bronze_patients"
)
spark.read.format("delta").load(f"{BRONZE_PATH}/encounters").createOrReplaceTempView(
    "bronze_encounters"
)
spark.read.format("delta").load(f"{BRONZE_PATH}/conditions").createOrReplaceTempView(
    "bronze_conditions"
)

# Verify row counts and sample data
spark.sql("SELECT COUNT(*) as patient_count FROM bronze_patients").show()
spark.sql("SELECT * FROM bronze_patients LIMIT 3").show(truncate=False)

What does this code do?

  • .createOrReplaceTempView() registers each Delta table as a named view in your Spark session. This lets you use familiar SQL syntax instead of the DataFrame API.
  • spark.sql() executes a SQL string and returns a DataFrame. The .show() call prints the results in a formatted table. Using truncate=False displays full column values without cutting them off.
  • Run the cell with Shift+Enter.

You should see a table showing patient_count = 1000, followed by 3 sample patient rows with all their columns including the _ingested_at timestamp at the end.

Query returning 0 rows or table not found?

  • If the table is not found, make sure you ran Cell 4 (the Bronze write) before this cell. Temp views only work if the underlying files exist.
  • If you restarted your Jupyter kernel, you need to re-run all cells from the top. Temp views are session-scoped and do not persist across kernel restarts.

Help me figure out why my Spark SQL query isn't returning data.

✔️ Awesome, I've got everything!

Great. Double check you have saved your notebook (press Cmd+S on macOS or Ctrl+S on Windows).

ⓧ I'd like to double check the full code

Here is everything you should have in your notebook so far (5 cells total):

# Cell 1: Initialize Spark with Delta Lake
import pyspark
from delta import *
from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, DateType, TimestampType
)
from pyspark.sql.functions import (
    col, year, month, to_date, to_timestamp, when, lit,
    count, avg, round as spark_round, current_timestamp,
    datediff, expr, desc
)

builder = (
    SparkSession.builder
    .appName("HealthcareLakehouse")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog"
    )
    .master("local[*]")
)

spark = configure_spark_with_delta_pip(builder).getOrCreate()
spark.sparkContext.setLogLevel("WARN")
print(f"Spark version: {spark.version}")
print("Delta Lake configured successfully!")

# Cell 2: Define paths
BRONZE_PATH = "/home/jovyan/work/data/bronze"
SILVER_PATH = "/home/jovyan/work/data/silver"
GOLD_PATH = "/home/jovyan/work/data/gold"
RAW_PATH = "/home/jovyan/work/data/raw"

# Cell 3: Read raw CSV files
patients_raw = spark.read.csv(
    f"{RAW_PATH}/patients.csv", header=True, inferSchema=True
)
encounters_raw = spark.read.csv(
    f"{RAW_PATH}/encounters.csv", header=True, inferSchema=True
)
conditions_raw = spark.read.csv(
    f"{RAW_PATH}/conditions.csv", header=True, inferSchema=True
)

print(f"Patients: {patients_raw.count()} rows")
print(f"Encounters: {encounters_raw.count()} rows")
print(f"Conditions: {conditions_raw.count()} rows")
patients_raw.printSchema()

# Cell 4: Write Bronze Delta tables
patients_bronze = patients_raw.withColumn("_ingested_at", current_timestamp())
encounters_bronze = encounters_raw.withColumn("_ingested_at", current_timestamp())
conditions_bronze = conditions_raw.withColumn("_ingested_at", current_timestamp())

patients_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/patients"
)
encounters_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/encounters"
)
conditions_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/conditions"
)
print("Bronze layer written successfully!")

# Cell 5: Verify Bronze with Spark SQL
spark.read.format("delta").load(f"{BRONZE_PATH}/patients").createOrReplaceTempView(
    "bronze_patients"
)
spark.read.format("delta").load(f"{BRONZE_PATH}/encounters").createOrReplaceTempView(
    "bronze_encounters"
)
spark.read.format("delta").load(f"{BRONZE_PATH}/conditions").createOrReplaceTempView(
    "bronze_conditions"
)

spark.sql("SELECT COUNT(*) as patient_count FROM bronze_patients").show()
spark.sql("SELECT * FROM bronze_patients LIMIT 3").show(truncate=False)

Your raw data is now persisted in Delta format with full ACID guarantees. Next up, you will clean and transform this Bronze data into a curated Silver layer with proper types, derived columns, and filtered invalid records.

Build the Silver Layer

Your Bronze layer captured raw data exactly as it arrived. Now you need to clean and standardize it so downstream analytics can trust the results.

The Silver layer is where you cast columns to proper data types, derive new columns like age and duration, rename fields to a clean convention, and filter out invalid records. This transforms messy ingestion data into a curated, analysis-ready format.

In this step, get ready to:
  • Clean and type-cast patient data with proper date handling and age derivation.
  • Clean and enrich encounter data with timestamps and calculated durations.
  • Write curated Silver Delta tables that filter out invalid records.
Clean and type-cast patient data

The Bronze patients table stores dates as plain strings and has no validation. You need proper DateType columns, a derived age column, and filters to remove impossible records like future birthdates.

  • In your notebook, add a new markdown cell below Cell 5 by clicking the + button, then paste this section header:
# %% [markdown]
# ## Silver Layer: Clean and Transform
  • Add a new code cell below the markdown cell and paste the patient cleaning transformation:
# %% Cell 6: Clean patients
patients_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/patients")
    .withColumn("birth_date", to_date(col("BIRTHDATE"), "yyyy-MM-dd"))
    .withColumn(
        "death_date",
        when(col("DEATHDATE") != "", to_date(col("DEATHDATE"), "yyyy-MM-dd"))
        .otherwise(lit(None).cast(DateType()))
    )
    .withColumn(
        "age",
        when(
            col("death_date").isNotNull(),
            (datediff(col("death_date"), col("birth_date")) / 365).cast("int")
        ).otherwise(
            (datediff(current_timestamp(), col("birth_date")) / 365).cast("int")
        )
    )
    .filter(col("birth_date") < current_timestamp())
    .filter(col("age") > 0)
    .select(
        col("Id").alias("patient_id"),
        "birth_date",
        "death_date",
        "age",
        col("GENDER").alias("gender"),
        col("RACE").alias("race"),
        col("ETHNICITY").alias("ethnicity"),
        col("CITY").alias("city"),
        col("STATE").alias("state"),
        col("ZIP").alias("zip_code"),
        col("HEALTHCARE_EXPENSES").cast(DoubleType()).alias("healthcare_expenses"),
        col("HEALTHCARE_COVERAGE").cast(DoubleType()).alias("healthcare_coverage"),
        col("INCOME").cast(DoubleType()).alias("income"),
    )
)

print(f"Silver patients: {patients_silver.count()} rows")
patients_silver.printSchema()

What does this code do?

This transformation reads from the Bronze Delta table and applies several cleaning operations in a chain:

  • to_date() converts BIRTHDATE and DEATHDATE strings into proper DateType columns. Empty death dates become null instead of empty strings.
  • The age column uses datediff() to calculate years between birth and death (or birth and now for living patients).
  • Two .filter() calls remove records with future birthdates or non-positive ages.
  • The .select() renames columns from UPPERCASE (raw CSV convention) to snake_case (analytics convention) and casts numeric columns to DoubleType.
  • Run Cell 6 by pressing Shift+Enter.
  • You should see a row count (slightly less than 1000 due to filtered records) and the new schema printed with proper types like DateType and DoubleType.

Seeing an AnalysisException?

If you see an error about columns not found, make sure you ran the Bronze layer cells (Cells 3-5) first. The Silver layer reads from the Bronze Delta tables, so they must exist.

If the error mentions BRONZE_PATH not being defined, re-run Cell 2 where the path constants are set.

Help me debug my Silver layer patient cleaning code

Clean and enrich encounter data

Encounter records contain start and stop times as ISO 8601 strings. You need proper timestamps, a calculated duration, and extracted year/month fields for filtering.

  • Add a new code cell below Cell 6 and paste the encounter cleaning transformation:
# %% Cell 7: Clean encounters
encounters_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/encounters")
    .withColumn("start_time", to_timestamp(col("START"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn("stop_time", to_timestamp(col("STOP"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn(
        "duration_minutes",
        (col("stop_time").cast("long") - col("start_time").cast("long")) / 60
    )
    .withColumn("encounter_year", year(col("start_time")))
    .withColumn("encounter_month", month(col("start_time")))
    .filter(col("start_time").isNotNull())
    .filter(col("duration_minutes") > 0)
    .select(
        col("Id").alias("encounter_id"),
        col("PATIENT").alias("patient_id"),
        "start_time",
        "stop_time",
        "duration_minutes",
        "encounter_year",
        "encounter_month",
        col("ENCOUNTERCLASS").alias("encounter_class"),
        col("DESCRIPTION").alias("description"),
        col("BASE_ENCOUNTER_COST").cast(DoubleType()).alias("base_cost"),
        col("TOTAL_CLAIM_COST").cast(DoubleType()).alias("total_cost"),
        col("PAYER_COVERAGE").cast(DoubleType()).alias("payer_coverage"),
    )
)

print(f"Silver encounters: {encounters_silver.count()} rows")

What does this code do?

  • to_timestamp() parses the ISO 8601 START and STOP strings into proper TimestampType columns that Spark can compare and filter.
  • duration_minutes is derived by casting both timestamps to seconds (long), subtracting, and dividing by 60.
  • encounter_year and encounter_month are extracted with Spark's year() and month() functions. These become useful for partitioning and filtering later.
  • Records with null start times or non-positive durations are filtered out as invalid.
  • Run Cell 7 by pressing Shift+Enter.
  • You should see a row count printed for the Silver encounters. The count may be slightly lower than the Bronze encounters count since invalid records are filtered.

Row count is zero?

A zero row count usually means the timestamp format pattern doesn't match the data. Check that the pattern yyyy-MM-dd'T'HH:mm:ss'Z' matches what your data generation script wrote. If all timestamps parse as null, the .filter(col("start_time").isNotNull()) removes every row.

Help me figure out why my encounter timestamps are all null

Write Silver Delta tables

With both DataFrames cleaned, you can persist them as Delta Lake tables in the Silver directory. This creates versioned, ACID-compliant tables that later steps can read, evolve, and query with time travel.

  • Add a new code cell below Cell 7 and paste the Silver write logic:
# %% Cell 8: Write Silver Delta tables
patients_silver.write.format("delta").mode("overwrite").save(
    f"{SILVER_PATH}/patients"
)
encounters_silver.write.format("delta").mode("overwrite").save(
    f"{SILVER_PATH}/encounters"
)
print("Silver layer written successfully!")

What does this code do?

Each .write.format("delta").mode("overwrite").save() call persists the cleaned DataFrame as a Delta table in the Silver directory. The overwrite mode replaces any existing data, making this cell safe to re-run.

Delta Lake writes a _delta_log directory alongside the Parquet data files. This transaction log is what enables schema evolution, MERGE upserts, and time travel in future steps.

  • Run Cell 8 by pressing Shift+Enter.
  • You should see "Silver layer written successfully!" printed in the output.

Seeing a NameError for patients_silver?

This means Cell 6 didn't run successfully. Go back and re-run Cell 6, then Cell 7, then Cell 8 in order. Jupyter cells depend on previous cells having completed.

Help me debug my Silver Delta table write

Why is the Silver row count lower than Bronze?

The Bronze layer stores everything exactly as ingested. The Silver layer applies quality filters: patients with impossible birthdates or non-positive ages are removed, and encounters with null timestamps or zero durations are dropped.

This is the medallion architecture in action. Bronze preserves raw data for auditing. Silver ensures only valid, well-typed records flow into analytics.

✔️ Awesome, I've got everything!

Great. Double check you've saved your notebook.

ⓧ I'd like to double check the full code

Here are all the cells you should have from this step (Cells 6-8), in addition to the Cells 1-5 you built previously:

# %% [markdown]
# ## Silver Layer: Clean and Transform

# %% Cell 6: Clean patients
patients_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/patients")
    .withColumn("birth_date", to_date(col("BIRTHDATE"), "yyyy-MM-dd"))
    .withColumn(
        "death_date",
        when(col("DEATHDATE") != "", to_date(col("DEATHDATE"), "yyyy-MM-dd"))
        .otherwise(lit(None).cast(DateType()))
    )
    .withColumn(
        "age",
        when(
            col("death_date").isNotNull(),
            (datediff(col("death_date"), col("birth_date")) / 365).cast("int")
        ).otherwise(
            (datediff(current_timestamp(), col("birth_date")) / 365).cast("int")
        )
    )
    .filter(col("birth_date") < current_timestamp())
    .filter(col("age") > 0)
    .select(
        col("Id").alias("patient_id"),
        "birth_date",
        "death_date",
        "age",
        col("GENDER").alias("gender"),
        col("RACE").alias("race"),
        col("ETHNICITY").alias("ethnicity"),
        col("CITY").alias("city"),
        col("STATE").alias("state"),
        col("ZIP").alias("zip_code"),
        col("HEALTHCARE_EXPENSES").cast(DoubleType()).alias("healthcare_expenses"),
        col("HEALTHCARE_COVERAGE").cast(DoubleType()).alias("healthcare_coverage"),
        col("INCOME").cast(DoubleType()).alias("income"),
    )
)

print(f"Silver patients: {patients_silver.count()} rows")
patients_silver.printSchema()

# %% Cell 7: Clean encounters
encounters_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/encounters")
    .withColumn("start_time", to_timestamp(col("START"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn("stop_time", to_timestamp(col("STOP"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn(
        "duration_minutes",
        (col("stop_time").cast("long") - col("start_time").cast("long")) / 60
    )
    .withColumn("encounter_year", year(col("start_time")))
    .withColumn("encounter_month", month(col("start_time")))
    .filter(col("start_time").isNotNull())
    .filter(col("duration_minutes") > 0)
    .select(
        col("Id").alias("encounter_id"),
        col("PATIENT").alias("patient_id"),
        "start_time",
        "stop_time",
        "duration_minutes",
        "encounter_year",
        "encounter_month",
        col("ENCOUNTERCLASS").alias("encounter_class"),
        col("DESCRIPTION").alias("description"),
        col("BASE_ENCOUNTER_COST").cast(DoubleType()).alias("base_cost"),
        col("TOTAL_CLAIM_COST").cast(DoubleType()).alias("total_cost"),
        col("PAYER_COVERAGE").cast(DoubleType()).alias("payer_coverage"),
    )
)

print(f"Silver encounters: {encounters_silver.count()} rows")

# %% Cell 8: Write Silver Delta tables
patients_silver.write.format("delta").mode("overwrite").save(
    f"{SILVER_PATH}/patients"
)
encounters_silver.write.format("delta").mode("overwrite").save(
    f"{SILVER_PATH}/encounters"
)
print("Silver layer written successfully!")

Your Silver layer is locked in. You have clean, properly typed patient and encounter tables ready for analytics. Next up, you'll build Gold aggregation tables and query them with Spark SQL to answer real healthcare questions.

Build the Gold Layer and Query with Spark SQL

In the previous step, you cleaned raw healthcare data into structured Silver tables with proper types, derived columns, and filtered invalid records. Now you have curated data ready for analysis.

Analysts do not want to write complex joins and aggregations every time they need a report. The Gold layer solves this by pre-computing summary tables optimized for business questions. In this step, you will build two Gold aggregation tables and query them with Spark SQL.

In this step, get ready to:
  • Create Gold aggregation tables for patient encounter summaries and condition prevalence.
  • Register Delta tables as temporary views for SQL queries.
  • Run analytical Spark SQL queries on the Gold layer.
Create Gold aggregation tables

The Gold layer contains pre-aggregated, analytics-ready tables. Each table answers a specific business question without requiring downstream consumers to repeat complex transformations.

  • In your notebook, add a new markdown cell and a code cell for Cell 9 by pasting the following code:
# %% [markdown]
# ## Gold Layer: Analytics

# %% Cell 9: Create Gold aggregation - Patient Encounter Summary
silver_encounters = spark.read.format("delta").load(f"{SILVER_PATH}/encounters")
silver_patients = spark.read.format("delta").load(f"{SILVER_PATH}/patients")

patient_encounter_summary = (
    silver_encounters
    .groupBy("patient_id")
    .agg(
        count("encounter_id").alias("total_encounters"),
        spark_round(avg("total_cost"), 2).alias("avg_encounter_cost"),
        spark_round(avg("duration_minutes"), 2).alias("avg_duration_minutes"),
    )
    .join(
        silver_patients.select("patient_id", "age", "gender", "city"),
        on="patient_id",
        how="left"
    )
)

patient_encounter_summary.write.format("delta").mode("overwrite").save(
    f"{GOLD_PATH}/patient_encounter_summary"
)

What does this code do?

This cell reads from your Silver tables and builds a per-patient summary.

  • `groupBy("patient_id")` collapses all encounters for a single patient into one row.
  • `count("encounter_id")` counts how many encounters each patient had.
  • `avg("total_cost")` and `avg("duration_minutes")` compute averages per patient.
  • The `.join()` enriches encounter stats with patient demographics (age, gender, city) from the Silver patients table.
  • The result is written to `data/gold/patient_encounter_summary` as a Delta table.
  • Run the cell.
  • You won't see visible output yet because this cell writes the table silently. The next chunk confirms it worked.
  • Add a new code cell for Cell 10 and paste the following code:
# %% Cell 10: Create Gold aggregation - Condition Prevalence
conditions_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/conditions")
    .select(
        col("CODE").alias("condition_code"),
        col("DESCRIPTION").alias("condition_name"),
        col("PATIENT").alias("patient_id"),
    )
)

condition_prevalence = (
    conditions_silver
    .groupBy("condition_code", "condition_name")
    .agg(
        count("patient_id").alias("total_occurrences"),
    )
    .orderBy(desc("total_occurrences"))
)

condition_prevalence.write.format("delta").mode("overwrite").save(
    f"{GOLD_PATH}/condition_prevalence"
)
print("Gold layer written successfully!")

What does this code do?

This cell builds a condition frequency table from the Bronze conditions data.

  • `conditions_silver` reads from Bronze (not Silver) because conditions were not separately cleaned into the Silver layer. It selects and renames columns for clarity.
  • `groupBy("condition_code", "condition_name")` groups all occurrences of the same condition.
  • `count("patient_id")` tallies how many times each condition appears across all patients.
  • `orderBy(desc("total_occurrences"))` ranks the most common conditions at the top.
  • Run Cell 10.
  • You should see `Gold layer written successfully!` printed in the output.

Not seeing the success message?

  • Check that `GOLD_PATH` is defined in Cell 2. If you restarted your kernel, re-run Cells 1 and 2 first.
  • Make sure you ran Cell 8 (Silver writes) before this cell. The patient encounter summary depends on Silver tables existing.

Why isn't my Gold layer writing successfully?

Query Gold tables with Spark SQL

With Gold tables written, you can register them as temporary views and run SQL queries directly. This mirrors how analysts interact with a lakehouse in production.

  • Add a new code cell for Cell 11 and paste the following code:
# %% Cell 11: Query Gold with Spark SQL
spark.read.format("delta").load(
    f"{GOLD_PATH}/condition_prevalence"
).createOrReplaceTempView("gold_conditions")

spark.read.format("delta").load(
    f"{GOLD_PATH}/patient_encounter_summary"
).createOrReplaceTempView("gold_patient_summary")

print("=== Top 10 Most Common Conditions ===")
spark.sql("""
    SELECT condition_name, total_occurrences
    FROM gold_conditions
    ORDER BY total_occurrences DESC
    LIMIT 10
""").show(truncate=False)

print("=== Average Cost by Encounter Class ===")
spark.read.format("delta").load(f"{SILVER_PATH}/encounters").createOrReplaceTempView(
    "silver_encounters"
)
spark.sql("""
    SELECT encounter_class,
           COUNT(*) as encounter_count,
           ROUND(AVG(total_cost), 2) as avg_cost,
           ROUND(AVG(duration_minutes), 2) as avg_duration_min
    FROM silver_encounters
    GROUP BY encounter_class
    ORDER BY avg_cost DESC
""").show()

print("=== Patient Cohort by Age Group ===")
spark.sql("""
    SELECT
        CASE
            WHEN age < 30 THEN '18-29'
            WHEN age < 50 THEN '30-49'
            WHEN age < 70 THEN '50-69'
            ELSE '70+'
        END as age_group,
        COUNT(*) as patient_count,
        ROUND(AVG(total_encounters), 1) as avg_encounters,
        ROUND(AVG(avg_encounter_cost), 2) as avg_cost
    FROM gold_patient_summary
    GROUP BY 1
    ORDER BY 1
""").show()

What do these queries do?

`createOrReplaceTempView` registers each Delta table as a named view that Spark SQL can query like a regular database table. The view exists for the lifetime of your SparkSession.

  • Query 1 finds the top 10 most common medical conditions by occurrence count.
  • Query 2 calculates average cost and duration per encounter class (ambulatory, emergency, inpatient, etc.).
  • Query 3 groups patients into age cohorts and shows how encounter frequency and costs vary by age.
  • Run Cell 11.
  • You should see three formatted tables printed in the output, each preceded by a section header like `=== Top 10 Most Common Conditions ===`.

Seeing an AnalysisException?

  • If the error mentions a table or view not found, make sure you ran Cells 9 and 10 first. The Gold tables must exist before you can register them as views.
  • If a column name is not found, confirm that your Silver tables have the expected columns by running `spark.read.format("delta").load(f"{SILVER_PATH}/encounters").columns` in a scratch cell.

Help me fix my Spark SQL query error.

✔️ Awesome, I've got everything!

Great. Double check you have saved your notebook.

ⓧ I'd like to double check the full code

Here is the full notebook code from Cell 1 through Cell 11:

# %% [markdown]
# # Healthcare Data Lakehouse with Delta Lake
# Bronze -> Silver -> Gold medallion architecture

# %% Cell 1: Initialize Spark with Delta Lake
import pyspark
from delta import *
from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, DateType, TimestampType
)
from pyspark.sql.functions import (
    col, year, month, to_date, to_timestamp, when, lit,
    count, avg, round as spark_round, current_timestamp,
    datediff, expr, desc
)

builder = (
    SparkSession.builder
    .appName("HealthcareLakehouse")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog"
    )
    .master("local[*]")
)

spark = configure_spark_with_delta_pip(builder).getOrCreate()
spark.sparkContext.setLogLevel("WARN")
print(f"Spark version: {spark.version}")
print("Delta Lake configured successfully!")

# %% Cell 2: Define paths
BRONZE_PATH = "/home/jovyan/work/data/bronze"
SILVER_PATH = "/home/jovyan/work/data/silver"
GOLD_PATH = "/home/jovyan/work/data/gold"
RAW_PATH = "/home/jovyan/work/data/raw"

# %% [markdown]
# ## Bronze Layer: Raw Ingestion

# %% Cell 3: Read raw CSV files
patients_raw = spark.read.csv(
    f"{RAW_PATH}/patients.csv", header=True, inferSchema=True
)
encounters_raw = spark.read.csv(
    f"{RAW_PATH}/encounters.csv", header=True, inferSchema=True
)
conditions_raw = spark.read.csv(
    f"{RAW_PATH}/conditions.csv", header=True, inferSchema=True
)

print(f"Patients: {patients_raw.count()} rows")
print(f"Encounters: {encounters_raw.count()} rows")
print(f"Conditions: {conditions_raw.count()} rows")
patients_raw.printSchema()

# %% Cell 4: Write Bronze Delta tables
patients_bronze = patients_raw.withColumn("_ingested_at", current_timestamp())
encounters_bronze = encounters_raw.withColumn("_ingested_at", current_timestamp())
conditions_bronze = conditions_raw.withColumn("_ingested_at", current_timestamp())

patients_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/patients"
)
encounters_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/encounters"
)
conditions_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/conditions"
)
print("Bronze layer written successfully!")

# %% Cell 5: Verify Bronze with Spark SQL
spark.read.format("delta").load(f"{BRONZE_PATH}/patients").createOrReplaceTempView(
    "bronze_patients"
)
spark.read.format("delta").load(f"{BRONZE_PATH}/encounters").createOrReplaceTempView(
    "bronze_encounters"
)
spark.read.format("delta").load(f"{BRONZE_PATH}/conditions").createOrReplaceTempView(
    "bronze_conditions"
)

spark.sql("SELECT COUNT(*) as patient_count FROM bronze_patients").show()
spark.sql("SELECT * FROM bronze_patients LIMIT 3").show(truncate=False)

# %% [markdown]
# ## Silver Layer: Clean and Transform

# %% Cell 6: Clean patients
patients_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/patients")
    .withColumn("birth_date", to_date(col("BIRTHDATE"), "yyyy-MM-dd"))
    .withColumn(
        "death_date",
        when(col("DEATHDATE") != "", to_date(col("DEATHDATE"), "yyyy-MM-dd"))
        .otherwise(lit(None).cast(DateType()))
    )
    .withColumn(
        "age",
        when(
            col("death_date").isNotNull(),
            (datediff(col("death_date"), col("birth_date")) / 365).cast("int")
        ).otherwise(
            (datediff(current_timestamp(), col("birth_date")) / 365).cast("int")
        )
    )
    .filter(col("birth_date") < current_timestamp())
    .filter(col("age") > 0)
    .select(
        col("Id").alias("patient_id"),
        "birth_date",
        "death_date",
        "age",
        col("GENDER").alias("gender"),
        col("RACE").alias("race"),
        col("ETHNICITY").alias("ethnicity"),
        col("CITY").alias("city"),
        col("STATE").alias("state"),
        col("ZIP").alias("zip_code"),
        col("HEALTHCARE_EXPENSES").cast(DoubleType()).alias("healthcare_expenses"),
        col("HEALTHCARE_COVERAGE").cast(DoubleType()).alias("healthcare_coverage"),
        col("INCOME").cast(DoubleType()).alias("income"),
    )
)

print(f"Silver patients: {patients_silver.count()} rows")
patients_silver.printSchema()

# %% Cell 7: Clean encounters
encounters_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/encounters")
    .withColumn("start_time", to_timestamp(col("START"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn("stop_time", to_timestamp(col("STOP"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn(
        "duration_minutes",
        (col("stop_time").cast("long") - col("start_time").cast("long")) / 60
    )
    .withColumn("encounter_year", year(col("start_time")))
    .withColumn("encounter_month", month(col("start_time")))
    .filter(col("start_time").isNotNull())
    .filter(col("duration_minutes") > 0)
    .select(
        col("Id").alias("encounter_id"),
        col("PATIENT").alias("patient_id"),
        "start_time",
        "stop_time",
        "duration_minutes",
        "encounter_year",
        "encounter_month",
        col("ENCOUNTERCLASS").alias("encounter_class"),
        col("DESCRIPTION").alias("description"),
        col("BASE_ENCOUNTER_COST").cast(DoubleType()).alias("base_cost"),
        col("TOTAL_CLAIM_COST").cast(DoubleType()).alias("total_cost"),
        col("PAYER_COVERAGE").cast(DoubleType()).alias("payer_coverage"),
    )
)

print(f"Silver encounters: {encounters_silver.count()} rows")

# %% Cell 8: Write Silver Delta tables
patients_silver.write.format("delta").mode("overwrite").save(
    f"{SILVER_PATH}/patients"
)
encounters_silver.write.format("delta").mode("overwrite").save(
    f"{SILVER_PATH}/encounters"
)
print("Silver layer written successfully!")

# %% [markdown]
# ## Gold Layer: Analytics

# %% Cell 9: Create Gold aggregation - Patient Encounter Summary
silver_encounters = spark.read.format("delta").load(f"{SILVER_PATH}/encounters")
silver_patients = spark.read.format("delta").load(f"{SILVER_PATH}/patients")

patient_encounter_summary = (
    silver_encounters
    .groupBy("patient_id")
    .agg(
        count("encounter_id").alias("total_encounters"),
        spark_round(avg("total_cost"), 2).alias("avg_encounter_cost"),
        spark_round(avg("duration_minutes"), 2).alias("avg_duration_minutes"),
    )
    .join(
        silver_patients.select("patient_id", "age", "gender", "city"),
        on="patient_id",
        how="left"
    )
)

patient_encounter_summary.write.format("delta").mode("overwrite").save(
    f"{GOLD_PATH}/patient_encounter_summary"
)

# %% Cell 10: Create Gold aggregation - Condition Prevalence
conditions_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/conditions")
    .select(
        col("CODE").alias("condition_code"),
        col("DESCRIPTION").alias("condition_name"),
        col("PATIENT").alias("patient_id"),
    )
)

condition_prevalence = (
    conditions_silver
    .groupBy("condition_code", "condition_name")
    .agg(
        count("patient_id").alias("total_occurrences"),
    )
    .orderBy(desc("total_occurrences"))
)

condition_prevalence.write.format("delta").mode("overwrite").save(
    f"{GOLD_PATH}/condition_prevalence"
)
print("Gold layer written successfully!")

# %% Cell 11: Query Gold with Spark SQL
spark.read.format("delta").load(
    f"{GOLD_PATH}/condition_prevalence"
).createOrReplaceTempView("gold_conditions")

spark.read.format("delta").load(
    f"{GOLD_PATH}/patient_encounter_summary"
).createOrReplaceTempView("gold_patient_summary")

print("=== Top 10 Most Common Conditions ===")
spark.sql("""
    SELECT condition_name, total_occurrences
    FROM gold_conditions
    ORDER BY total_occurrences DESC
    LIMIT 10
""").show(truncate=False)

print("=== Average Cost by Encounter Class ===")
spark.read.format("delta").load(f"{SILVER_PATH}/encounters").createOrReplaceTempView(
    "silver_encounters"
)
spark.sql("""
    SELECT encounter_class,
           COUNT(*) as encounter_count,
           ROUND(AVG(total_cost), 2) as avg_cost,
           ROUND(AVG(duration_minutes), 2) as avg_duration_min
    FROM silver_encounters
    GROUP BY encounter_class
    ORDER BY avg_cost DESC
""").show()

print("=== Patient Cohort by Age Group ===")
spark.sql("""
    SELECT
        CASE
            WHEN age < 30 THEN '18-29'
            WHEN age < 50 THEN '30-49'
            WHEN age < 70 THEN '50-69'
            ELSE '70+'
        END as age_group,
        COUNT(*) as patient_count,
        ROUND(AVG(total_encounters), 1) as avg_encounters,
        ROUND(AVG(avg_encounter_cost), 2) as avg_cost
    FROM gold_patient_summary
    GROUP BY 1
    ORDER BY 1
""").show()

Your Gold layer is complete. You now have a fully functional medallion architecture with Bronze, Silver, and Gold layers. Next up, you will explore three powerful Delta Lake features: schema evolution, upserts, and time travel.

Schema Evolution, Upserts, and Time Travel

Your lakehouse has a complete medallion architecture with Bronze, Silver, and Gold layers producing real analytics. Now it is time to prove that your Delta Lake tables can handle real-world change.

In production, data schemas evolve as business requirements shift. Patient records get updated, and auditors need to see what the data looked like before those updates. This step demonstrates three features that make Delta Lake production-ready: schema evolution, MERGE upserts, and time travel.

In this step, get ready to:
  • Add a new column to an existing Delta table without rewriting it.
  • Update specific rows using a MERGE upsert operation.
  • Query previous table versions using time travel.
Demonstrate schema evolution

In a traditional data warehouse, adding a column means rewriting the entire table or running a costly ALTER TABLE migration. Delta Lake supports schema evolution natively. You can add new columns on write and the existing data remains untouched.

  • Add a new cell in your notebook (Cell 12) to compute a risk_score column and write it back to the Silver patients table with schema evolution enabled:
# %% Cell 12: Schema Evolution - Add risk_score column
import random

silver_patients_df = spark.read.format("delta").load(f"{SILVER_PATH}/patients")
patients_with_risk = silver_patients_df.withColumn(
    "risk_score",
    spark_round(
        when(col("age") > 65, lit(0.7) + (col("age") - 65) * 0.005)
        .when(col("age") > 50, lit(0.4) + (col("age") - 50) * 0.02)
        .otherwise(lit(0.1) + col("age") * 0.005),
        3
    )
)

patients_with_risk.write.format("delta").mode("overwrite").option(
    "mergeSchema", "true"
).save(f"{SILVER_PATH}/patients")

print("Schema evolved! New schema:")
spark.read.format("delta").load(f"{SILVER_PATH}/patients").printSchema()

What does this code do?

This cell reads the existing Silver patients table, computes a risk_score based on patient age (higher age means higher risk), and writes the result back.

The key line is .option("mergeSchema", "true"). This tells Delta Lake to automatically add any new columns in the DataFrame to the existing table schema. Without this option, Delta Lake would reject the write because the schema does not match.

  • Run Cell 12.
  • Check the printed schema output. You should see risk_score listed as a new column at the end of the schema.

Schema not showing risk_score?

Make sure mergeSchema is spelled exactly as shown (camelCase, no spaces). If you see a schema mismatch error, double-check that you included .option("mergeSchema", "true") before the .save() call.

help me fix a schema evolution error in Delta Lake

Perform an upsert with MERGE

An upsert combines "update" and "insert" into a single atomic operation. If a record already exists (matched by key), update it. If it does not exist, insert it. Delta Lake provides this through the MERGE operation.

  • Add Cell 13 to simulate updated patient records and merge them into the Silver table:
# %% Cell 13: Upsert with MERGE
updated_patients_data = (
    spark.read.format("delta").load(f"{SILVER_PATH}/patients")
    .limit(50)
    .withColumn("city", lit("Updated City"))
    .withColumn("risk_score", col("risk_score") + 0.1)
)

delta_patients = DeltaTable.forPath(spark, f"{SILVER_PATH}/patients")

delta_patients.alias("target").merge(
    updated_patients_data.alias("source"),
    "target.patient_id = source.patient_id"
).whenMatchedUpdate(
    set={
        "city": "source.city",
        "risk_score": "source.risk_score",
    }
).whenNotMatchedInsertAll().execute()

print("MERGE upsert complete!")
print("Verify updated rows:")
spark.read.format("delta").load(f"{SILVER_PATH}/patients").filter(
    col("city") == "Updated City"
).select("patient_id", "city", "risk_score").show(5)

What does this code do?

First, it creates a "source" DataFrame by taking 50 existing patients and changing their city to Updated City and incrementing their risk_score by 0.1. This simulates receiving updated records from an upstream system.

Then DeltaTable.forPath() opens the Silver patients table as a DeltaTable object. The .merge() call matches source and target rows on patient_id. For matches, only city and risk_score are updated. Unmatched rows are inserted as new records.

  • Run Cell 13.
  • You should see 5 rows printed with city set to Updated City and risk_score values slightly higher than the original.

Seeing an AnalysisException?

If you see an error about the table not being a Delta table, make sure Cell 12 ran successfully first. The MERGE operation requires the table to already exist as a valid Delta table at the path.

help me debug a MERGE upsert error in my Delta Lake patients table

Query historical versions with time travel

Every write to a Delta table creates a new version. Time travel lets you read any previous version by number or timestamp. This is invaluable for auditing changes and debugging data issues.

  • Add Cell 14 to compare the current table state against its previous versions:
# %% Cell 14: Time Travel
print("=== Current version (after MERGE) ===")
current_df = spark.read.format("delta").load(f"{SILVER_PATH}/patients")
print(f"Total rows: {current_df.count()}")
print(f"Rows with 'Updated City': {current_df.filter(col('city') == 'Updated City').count()}")

print("\n=== Version 0 (original Silver write) ===")
version_0 = spark.read.format("delta").option("versionAsOf", 0).load(
    f"{SILVER_PATH}/patients"
)
print(f"Total rows: {version_0.count()}")
print(f"Columns: {version_0.columns}")
print(f"Has risk_score: {'risk_score' in version_0.columns}")

print("\n=== Version 1 (after schema evolution) ===")
version_1 = spark.read.format("delta").option("versionAsOf", 1).load(
    f"{SILVER_PATH}/patients"
)
print(f"Total rows: {version_1.count()}")
print(f"Has risk_score: {'risk_score' in version_1.columns}")
print(f"Rows with 'Updated City': {version_1.filter(col('city') == 'Updated City').count()}")

print("\n=== Delta Table History ===")
delta_patients.history().select(
    "version", "timestamp", "operation", "operationMetrics"
).show(truncate=False)

What does this code do?

This cell reads three versions of the same table. The current version (after the MERGE) has rows with Updated City and the risk_score column.

Version 0 is the original Silver write from Step 3. It has no risk_score column at all. Version 1 is after schema evolution added risk_score, but before the MERGE updated any cities.

The .option("versionAsOf", 0) syntax is all it takes to read a historical snapshot. The .history() call shows the full audit log of every operation performed on the table.

  • Run Cell 14.
  • Verify that version 0 reports Has risk_score: False and version 1 reports Has risk_score: True with 0 rows having Updated City.
  • Check the Delta Table History output. You should see three versions: WRITE (version 0), WRITE (version 1, the schema evolution overwrite), and MERGE (version 2).

Time travel returning unexpected results?

If version 0 shows risk_score as present, you may have re-run cells out of order. Each overwrite creates a new version, so running Cell 8 (Silver write) again would reset your version numbering. Restart your kernel and run all cells in sequence from Cell 1.

help me understand why my Delta Lake time travel versions are not matching expectations

  • Add Cell 15 to stop your Spark session:
# %% Cell 15: Stop Spark
spark.stop()
print("Spark session stopped. Lakehouse project complete!")
  • Run Cell 15 to cleanly shut down Spark.

✔️ Awesome, I've got everything!

Great. Double check you have saved your notebook file.

ⓧ I'd like to double check the full code

Here is the full notebook from Cell 1 through Cell 15:

# %% [markdown]
# # Healthcare Data Lakehouse with Delta Lake
# Bronze -> Silver -> Gold medallion architecture

# %% Cell 1: Initialize Spark with Delta Lake
import pyspark
from delta import *
from delta.tables import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, DateType, TimestampType
)
from pyspark.sql.functions import (
    col, year, month, to_date, to_timestamp, when, lit,
    count, avg, round as spark_round, current_timestamp,
    datediff, expr, desc
)

builder = (
    SparkSession.builder
    .appName("HealthcareLakehouse")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.spark_catalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog"
    )
    .master("local[*]")
)

spark = configure_spark_with_delta_pip(builder).getOrCreate()
spark.sparkContext.setLogLevel("WARN")
print(f"Spark version: {spark.version}")
print("Delta Lake configured successfully!")

# %% Cell 2: Define paths
BRONZE_PATH = "/home/jovyan/work/data/bronze"
SILVER_PATH = "/home/jovyan/work/data/silver"
GOLD_PATH = "/home/jovyan/work/data/gold"
RAW_PATH = "/home/jovyan/work/data/raw"

# %% [markdown]
# ## Bronze Layer: Raw Ingestion

# %% Cell 3: Read raw CSV files
patients_raw = spark.read.csv(
    f"{RAW_PATH}/patients.csv", header=True, inferSchema=True
)
encounters_raw = spark.read.csv(
    f"{RAW_PATH}/encounters.csv", header=True, inferSchema=True
)
conditions_raw = spark.read.csv(
    f"{RAW_PATH}/conditions.csv", header=True, inferSchema=True
)

print(f"Patients: {patients_raw.count()} rows")
print(f"Encounters: {encounters_raw.count()} rows")
print(f"Conditions: {conditions_raw.count()} rows")
patients_raw.printSchema()

# %% Cell 4: Write Bronze Delta tables
patients_bronze = patients_raw.withColumn("_ingested_at", current_timestamp())
encounters_bronze = encounters_raw.withColumn("_ingested_at", current_timestamp())
conditions_bronze = conditions_raw.withColumn("_ingested_at", current_timestamp())

patients_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/patients"
)
encounters_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/encounters"
)
conditions_bronze.write.format("delta").mode("overwrite").save(
    f"{BRONZE_PATH}/conditions"
)
print("Bronze layer written successfully!")

# %% Cell 5: Verify Bronze with Spark SQL
spark.read.format("delta").load(f"{BRONZE_PATH}/patients").createOrReplaceTempView(
    "bronze_patients"
)
spark.read.format("delta").load(f"{BRONZE_PATH}/encounters").createOrReplaceTempView(
    "bronze_encounters"
)
spark.read.format("delta").load(f"{BRONZE_PATH}/conditions").createOrReplaceTempView(
    "bronze_conditions"
)

spark.sql("SELECT COUNT(*) as patient_count FROM bronze_patients").show()
spark.sql("SELECT * FROM bronze_patients LIMIT 3").show(truncate=False)

# %% [markdown]
# ## Silver Layer: Clean and Transform

# %% Cell 6: Clean patients
patients_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/patients")
    .withColumn("birth_date", to_date(col("BIRTHDATE"), "yyyy-MM-dd"))
    .withColumn(
        "death_date",
        when(col("DEATHDATE") != "", to_date(col("DEATHDATE"), "yyyy-MM-dd"))
        .otherwise(lit(None).cast(DateType()))
    )
    .withColumn(
        "age",
        when(
            col("death_date").isNotNull(),
            (datediff(col("death_date"), col("birth_date")) / 365).cast("int")
        ).otherwise(
            (datediff(current_timestamp(), col("birth_date")) / 365).cast("int")
        )
    )
    .filter(col("birth_date") < current_timestamp())
    .filter(col("age") > 0)
    .select(
        col("Id").alias("patient_id"),
        "birth_date",
        "death_date",
        "age",
        col("GENDER").alias("gender"),
        col("RACE").alias("race"),
        col("ETHNICITY").alias("ethnicity"),
        col("CITY").alias("city"),
        col("STATE").alias("state"),
        col("ZIP").alias("zip_code"),
        col("HEALTHCARE_EXPENSES").cast(DoubleType()).alias("healthcare_expenses"),
        col("HEALTHCARE_COVERAGE").cast(DoubleType()).alias("healthcare_coverage"),
        col("INCOME").cast(DoubleType()).alias("income"),
    )
)

print(f"Silver patients: {patients_silver.count()} rows")
patients_silver.printSchema()

# %% Cell 7: Clean encounters
encounters_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/encounters")
    .withColumn("start_time", to_timestamp(col("START"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn("stop_time", to_timestamp(col("STOP"), "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn(
        "duration_minutes",
        (col("stop_time").cast("long") - col("start_time").cast("long")) / 60
    )
    .withColumn("encounter_year", year(col("start_time")))
    .withColumn("encounter_month", month(col("start_time")))
    .filter(col("start_time").isNotNull())
    .filter(col("duration_minutes") > 0)
    .select(
        col("Id").alias("encounter_id"),
        col("PATIENT").alias("patient_id"),
        "start_time",
        "stop_time",
        "duration_minutes",
        "encounter_year",
        "encounter_month",
        col("ENCOUNTERCLASS").alias("encounter_class"),
        col("DESCRIPTION").alias("description"),
        col("BASE_ENCOUNTER_COST").cast(DoubleType()).alias("base_cost"),
        col("TOTAL_CLAIM_COST").cast(DoubleType()).alias("total_cost"),
        col("PAYER_COVERAGE").cast(DoubleType()).alias("payer_coverage"),
    )
)

print(f"Silver encounters: {encounters_silver.count()} rows")

# %% Cell 8: Write Silver Delta tables
patients_silver.write.format("delta").mode("overwrite").save(
    f"{SILVER_PATH}/patients"
)
encounters_silver.write.format("delta").mode("overwrite").save(
    f"{SILVER_PATH}/encounters"
)
print("Silver layer written successfully!")

# %% [markdown]
# ## Gold Layer: Analytics

# %% Cell 9: Create Gold aggregation - Patient Encounter Summary
silver_encounters = spark.read.format("delta").load(f"{SILVER_PATH}/encounters")
silver_patients = spark.read.format("delta").load(f"{SILVER_PATH}/patients")

patient_encounter_summary = (
    silver_encounters
    .groupBy("patient_id")
    .agg(
        count("encounter_id").alias("total_encounters"),
        spark_round(avg("total_cost"), 2).alias("avg_encounter_cost"),
        spark_round(avg("duration_minutes"), 2).alias("avg_duration_minutes"),
    )
    .join(
        silver_patients.select("patient_id", "age", "gender", "city"),
        on="patient_id",
        how="left"
    )
)

patient_encounter_summary.write.format("delta").mode("overwrite").save(
    f"{GOLD_PATH}/patient_encounter_summary"
)

# %% Cell 10: Create Gold aggregation - Condition Prevalence
conditions_silver = (
    spark.read.format("delta").load(f"{BRONZE_PATH}/conditions")
    .select(
        col("CODE").alias("condition_code"),
        col("DESCRIPTION").alias("condition_name"),
        col("PATIENT").alias("patient_id"),
    )
)

condition_prevalence = (
    conditions_silver
    .groupBy("condition_code", "condition_name")
    .agg(
        count("patient_id").alias("total_occurrences"),
    )
    .orderBy(desc("total_occurrences"))
)

condition_prevalence.write.format("delta").mode("overwrite").save(
    f"{GOLD_PATH}/condition_prevalence"
)
print("Gold layer written successfully!")

# %% Cell 11: Query Gold with Spark SQL
spark.read.format("delta").load(
    f"{GOLD_PATH}/condition_prevalence"
).createOrReplaceTempView("gold_conditions")

spark.read.format("delta").load(
    f"{GOLD_PATH}/patient_encounter_summary"
).createOrReplaceTempView("gold_patient_summary")

print("=== Top 10 Most Common Conditions ===")
spark.sql("""
    SELECT condition_name, total_occurrences
    FROM gold_conditions
    ORDER BY total_occurrences DESC
    LIMIT 10
""").show(truncate=False)

print("=== Average Cost by Encounter Class ===")
spark.read.format("delta").load(f"{SILVER_PATH}/encounters").createOrReplaceTempView(
    "silver_encounters"
)
spark.sql("""
    SELECT encounter_class,
           COUNT(*) as encounter_count,
           ROUND(AVG(total_cost), 2) as avg_cost,
           ROUND(AVG(duration_minutes), 2) as avg_duration_min
    FROM silver_encounters
    GROUP BY encounter_class
    ORDER BY avg_cost DESC
""").show()

print("=== Patient Cohort by Age Group ===")
spark.sql("""
    SELECT
        CASE
            WHEN age < 30 THEN '18-29'
            WHEN age < 50 THEN '30-49'
            WHEN age < 70 THEN '50-69'
            ELSE '70+'
        END as age_group,
        COUNT(*) as patient_count,
        ROUND(AVG(total_encounters), 1) as avg_encounters,
        ROUND(AVG(avg_encounter_cost), 2) as avg_cost
    FROM gold_patient_summary
    GROUP BY 1
    ORDER BY 1
""").show()

# %% [markdown]
# ## Schema Evolution, Upserts, and Time Travel

# %% Cell 12: Schema Evolution - Add risk_score column
import random

silver_patients_df = spark.read.format("delta").load(f"{SILVER_PATH}/patients")
patients_with_risk = silver_patients_df.withColumn(
    "risk_score",
    spark_round(
        when(col("age") > 65, lit(0.7) + (col("age") - 65) * 0.005)
        .when(col("age") > 50, lit(0.4) + (col("age") - 50) * 0.02)
        .otherwise(lit(0.1) + col("age") * 0.005),
        3
    )
)

patients_with_risk.write.format("delta").mode("overwrite").option(
    "mergeSchema", "true"
).save(f"{SILVER_PATH}/patients")

print("Schema evolved! New schema:")
spark.read.format("delta").load(f"{SILVER_PATH}/patients").printSchema()

# %% Cell 13: Upsert with MERGE
updated_patients_data = (
    spark.read.format("delta").load(f"{SILVER_PATH}/patients")
    .limit(50)
    .withColumn("city", lit("Updated City"))
    .withColumn("risk_score", col("risk_score") + 0.1)
)

delta_patients = DeltaTable.forPath(spark, f"{SILVER_PATH}/patients")

delta_patients.alias("target").merge(
    updated_patients_data.alias("source"),
    "target.patient_id = source.patient_id"
).whenMatchedUpdate(
    set={
        "city": "source.city",
        "risk_score": "source.risk_score",
    }
).whenNotMatchedInsertAll().execute()

print("MERGE upsert complete!")
print("Verify updated rows:")
spark.read.format("delta").load(f"{SILVER_PATH}/patients").filter(
    col("city") == "Updated City"
).select("patient_id", "city", "risk_score").show(5)

# %% Cell 14: Time Travel
print("=== Current version (after MERGE) ===")
current_df = spark.read.format("delta").load(f"{SILVER_PATH}/patients")
print(f"Total rows: {current_df.count()}")
print(f"Rows with 'Updated City': {current_df.filter(col('city') == 'Updated City').count()}")

print("\n=== Version 0 (original Silver write) ===")
version_0 = spark.read.format("delta").option("versionAsOf", 0).load(
    f"{SILVER_PATH}/patients"
)
print(f"Total rows: {version_0.count()}")
print(f"Columns: {version_0.columns}")
print(f"Has risk_score: {'risk_score' in version_0.columns}")

print("\n=== Version 1 (after schema evolution) ===")
version_1 = spark.read.format("delta").option("versionAsOf", 1).load(
    f"{SILVER_PATH}/patients"
)
print(f"Total rows: {version_1.count()}")
print(f"Has risk_score: {'risk_score' in version_1.columns}")
print(f"Rows with 'Updated City': {version_1.filter(col('city') == 'Updated City').count()}")

print("\n=== Delta Table History ===")
delta_patients.history().select(
    "version", "timestamp", "operation", "operationMetrics"
).show(truncate=False)

# %% Cell 15: Stop Spark
spark.stop()
print("Spark session stopped. Lakehouse project complete!")

Secret mission

Add Partitioning and Measure Performance

Your Silver encounters table works, but it scans every row for every query. In production, that means wasted compute on millions of records. Can you repartition the table and prove the performance gain with Spark's query planner?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using Docker, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker container (healthcare-lakehouse) running PySpark, Delta Lake, and Jupyter Notebook.
  • Local project folder (healthcare-lakehouse/) with data files, notebooks, and scripts (a few MB total).

Keep everything running

No action needed. Choose this if you're still actively building or want to experiment further with your lakehouse.

  • Your Docker container stays running and Jupyter Notebook remains accessible at http://localhost:8888.
  • All Delta tables, notebooks, and raw data remain on your machine.
  • The container uses minimal system resources when idle.

Pause - I'll come back to this later

Shut down running processes to free up memory while keeping all your files intact.

  • Open your terminal and navigate to your healthcare-lakehouse folder.
  • Stop the Docker container by running:
docker-compose stop
  • Your data, notebooks, and configuration files remain on disk.
  • When you're ready to resume, run docker-compose start from the same folder to pick up where you left off.

Delete - I don't want to use this again

Remove all project resources and start fresh.

  • Open your terminal and navigate to your healthcare-lakehouse folder.
  • Stop and remove the Docker container, network, and volumes by running:
docker-compose down -v
  • Navigate up one level and delete the project folder.

macOS

cd ~/Desktop
rm -rf healthcare-lakehouse

Windows

cd ~\Desktop
rmdir /s /q healthcare-lakehouse
  • This removes all Delta tables, notebooks, raw CSV files, and Docker configuration from your machine.

Nice Work!

Nice Work!

You just built a fully Dockerized healthcare data lakehouse from scratch. Your project processes synthetic patient data through a complete medallion architecture using Delta Lake and PySpark.

You've learned how to:

  • Build a portable Docker environment running PySpark, Delta Lake, and Jupyter Notebook for local data engineering.
  • Process data through a complete medallion architecture (Bronze, Silver, Gold) with raw ingestion, cleaning, type casting, and analytics-ready aggregations.
  • Use advanced Delta Lake features including schema evolution, MERGE upserts, and time travel to query historical versions of your data.
  • Secret Mission: Added date-based partitioning to the Silver layer and measured query performance improvement using partition pruning.

Ready to quiz yourself?