Implementing SCD Type 2 in PySpark and Delta Lake

Why this is more involved than Type 1

If you have already gone through implementing SCD Type 1, you know that pattern comes down to a fairly simple MERGE that updates matching rows and inserts new ones. Type 2 is genuinely more involved, because instead of updating a row in place, you need to close out the old row and insert a brand new row, while keeping both of them queryable with clear date ranges showing when each version was actually accurate.

Let's build this with the same customer scenario, so it is easy to compare directly against how Type 1 handled it.

Step 1, setting up the target table with tracking columns

For Type 2, our Dim_Customer table needs a few extra columns beyond just the business data, specifically to track history properly.

existing_data = [
    (501, "Rahul Mehta", "Bengaluru", "2024-01-01", None, True),
    (502, "Anita Sharma", "Pune", "2023-06-15", None, True),
]

columns = ["customer_id", "customer_name", "city", "effective_date", "end_date", "is_current"]

df_existing = spark.createDataFrame(existing_data, columns)

df_existing.write.format("delta") \
    .mode("overwrite") \
    .save("/mnt/datalake/gold/dim_customer_scd2/")

The effective_date marks when this version became true, end_date marks when it stopped being true, and is_current is a simple flag showing which row currently represents each customer's latest state. A null end_date means this row is still active.

Step 2, bring in today's updates

Same scenario as before. Rahul moved from Bengaluru to Pune, and there is a new customer, Priya.

incoming_data = [
    (501, "Rahul Mehta", "Pune"),
    (504, "Priya Nair", "Chennai")
]

incoming_columns = ["customer_id", "customer_name", "city"]

df_incoming = spark.createDataFrame(incoming_data, incoming_columns) \
    .withColumn("effective_date", F.current_date())

Notice Anita is not in this batch, so nothing should happen to her record.

Step 3, the actual SCD Type 2 logic

This needs two separate operations. First, close out any existing current row where the incoming data shows a genuine change. Second, insert new rows for both the changed customers and any brand new ones.

from delta.tables import DeltaTable
from pyspark.sql import functions as F

delta_table = DeltaTable.forPath(spark, "/mnt/datalake/gold/dim_customer_scd2/")

# Step 3a: identify which incoming records represent an actual change
df_current = delta_table.toDF().filter("is_current = true")

df_changes = df_incoming.alias("source").join(
    df_current.alias("target"),
    "customer_id",
    "left"
).filter(
    (F.col("target.customer_id").isNull()) |   # brand new customer
    (F.col("source.city") != F.col("target.city"))  # existing customer, value changed
).select("source.*")

# Step 3b: close out old current rows for customers that changed
delta_table.alias("target").merge(
    df_changes.alias("source"),
    "target.customer_id = source.customer_id AND target.is_current = true"
).whenMatchedUpdate(set={
    "end_date": "source.effective_date",
    "is_current": "false"
}).execute()

# Step 3c: insert new current rows
df_new_rows = df_changes.withColumn("end_date", F.lit(None).cast("date")) \
                         .withColumn("is_current", F.lit(True))

df_new_rows.write.format("delta").mode("append").save("/mnt/datalake/gold/dim_customer_scd2/")

Notice the join in step 3a is doing important work here, it filters df_incoming down to only genuinely new or changed records, so we are not accidentally creating a new row for Anita, whose data did not change at all.

Step 4, checking the result

spark.read.format("delta") \
    .load("/mnt/datalake/gold/dim_customer_scd2/") \
    .orderBy("customer_id", "effective_date") \
    .show()

Rahul now has two rows, the old Bengaluru version correctly closed out, and a new Pune version marked current. Anita is completely untouched since nothing changed for her. Priya was inserted as a fresh current row.

Querying this correctly

The part that trips people up after implementing this is that querying a Type 2 table is not as simple as it was with Type 1. If you just want the latest state of every customer, filter on is_current.

spark.sql("""
SELECT * FROM dim_customer_scd2 WHERE is_current = true
""")

But if you are joining this dimension to a fact table for historical reporting, like sales, you need to join based on the sale's date falling within the correct effective date range, not just the customer_id, to get the historically accurate version of that customer's data.

SELECT
    f.sale_id,
    f.sale_date,
    d.city
FROM Fact_Sales f
JOIN Dim_Customer_SCD2 d
    ON f.customer_id = d.customer_id
    AND f.sale_date >= d.effective_date
    AND (f.sale_date < d.end_date OR d.end_date IS NULL)

This is the query pattern that actually delivers the value SCD Type 2 was built for, correctly attributing a 2024 sale to Rahul's Bengaluru version, and a 2026 sale to his Pune version, rather than incorrectly showing all his historical sales under his current city.

A simpler alternative using Delta Lake's built in change tracking

For teams wanting to avoid writing this join and filter logic manually every time, Delta Lake also supports Change Data Feed, which can simplify tracking what changed between versions of a table, though implementing full SCD Type 2 logic with proper effective dating still generally requires this kind of explicit MERGE based approach for the dimension table itself. Change Data Feed is more commonly used for propagating changes downstream to other systems, rather than replacing the SCD Type 2 pattern shown here.

Interview angle

If asked to implement or explain SCD Type 2, being able to walk through why closing out the old row and inserting a new row are two separate operations, rather than a single update, shows real hands on understanding. It also helps to mention how querying a Type 2 table differs from a Type 1 table, specifically the need to join on effective date ranges for historical accuracy, since this is often where people who only understand the concept superficially get caught out by a follow up question.

Quick recap

Implementing SCD Type 2 in PySpark and Delta Lake requires identifying which incoming records represent genuine changes, closing out the corresponding current row by setting its end_date and is_current flag, and inserting a fresh row for the new current state. Unlike Type 1, querying this table for historical reporting requires joining on effective date ranges rather than just the customer_id, which is what actually delivers the historical accuracy this pattern is built to preserve.