SCD Type 2 Implementation in PySpark: A Complete Production Guide

Slowly Changing Dimension Type 2 (SCD-2) is a data warehousing strategy for tracking the full history of changes to dimension records over time. Unlike SCD-1 (which overwrites old values), SCD-2 preserves every historical version by inserting a new row for each change, using metadata columns like start_date, end_date, and is_current to differentiate between active and retired records.

SCD-2 is one of the most frequently asked scenarios in Data Engineering interviews and is critical in production lakehouse pipelines built on Delta Lake, Iceberg, or Hudi.

Below is the definitive, step-by-step implementation and diagnostic playbook using PySpark and Delta Lake MERGE.

SCD Type 2 Architecture Flow

The following diagram illustrates the end-to-end SCD-2 merge process: incoming source records flow through a Spark ETL process that performs INSERT, UPDATE, and RETAIN operations against the existing dimension table.

Step 1: Understand the Dimension Table Schema

An SCD-2 dimension table always requires these metadata columns in addition to the business attributes:

+----------------+-------------+------------+---------+------------+------------+------------+
| surrogate_key  | customer_id |   name     |  city   | start_date |  end_date  | is_current |
+----------------+-------------+------------+---------+------------+------------+------------+
|       1        |    C101     | Amit Kumar | Mumbai  | 2024-01-15 | 9999-12-31 |    true    |
|       2        |    C102     | Priya Rao  | Pune    | 2024-03-01 | 9999-12-31 |    true    |
|       3        |    C103     | Mukesh Sen | Delhi   | 2024-06-10 | 9999-12-31 |    true    |
+----------------+-------------+------------+---------+------------+------------+------------+

Column Definitions:

• surrogate_key: A system-generated unique identifier for each version of a row. This is NOT the business key.
• customer_id: The natural/business key — the real-world identifier that does not change even when attributes change.
• name, city: Business attributes that may change over time.
• start_date: The date this version of the record became effective.
• end_date: The date this version was retired. Active rows have 9999-12-31 (a far-future sentinel date).
• is_current: A boolean flag. true means this is the latest active version. false means this row is a historical snapshot.

Step 2: Set Up the Existing Dimension Table (Delta Lake)

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *
from delta.tables import DeltaTable

spark = SparkSession.builder \
    .appName("SCD-Type-2-Implementation") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Define schema for the dimension table
dim_schema = StructType([
    StructField("surrogate_key", LongType(), False),
    StructField("customer_id", StringType(), False),
    StructField("name", StringType(), True),
    StructField("city", StringType(), True),
    StructField("start_date", DateType(), False),
    StructField("end_date", DateType(), False),
    StructField("is_current", BooleanType(), False)
])

# Create sample existing dimension data
existing_data = [
    (1, "C101", "Amit Kumar", "Mumbai",  "2024-01-15", "9999-12-31", True),
    (2, "C102", "Priya Rao",  "Pune",    "2024-03-01", "9999-12-31", True),
    (3, "C103", "Mukesh Sen", "Delhi",   "2024-06-10", "9999-12-31", True),
]

dim_df = spark.createDataFrame(existing_data, 
    ["surrogate_key", "customer_id", "name", "city", "start_date", "end_date", "is_current"])

dim_df = dim_df.withColumn("start_date", F.to_date("start_date")) \
               .withColumn("end_date", F.to_date("end_date"))

# Write as a Delta table
DIMENSION_PATH = "/data/warehouse/dim_customer"
dim_df.write.format("delta").mode("overwrite").save(DIMENSION_PATH)

print("✅ Dimension table created successfully.")
dim_df.show(truncate=False)

Step 3: Simulate Incoming Source Records (CDC Feed)

In production, these records arrive from a Change Data Capture (CDC) system, a Kafka stream, or a daily extract from the transactional database. The source contains only the current state of changed records — it does NOT contain SCD metadata.

# Incoming source records for today's batch
incoming_data = [
    ("C101", "Amit Kumar", "Bangalore"),  # ⬆️ CHANGED: Amit moved from Mumbai → Bangalore
    ("C103", "Mukesh Sen", "Delhi"),      # ✅ UNCHANGED: Same city, no action needed
    ("C104", "Neha Gupta", "Hyderabad"),  # 🆕 NEW: Brand new customer, not in dimension
]

source_df = spark.createDataFrame(incoming_data, ["customer_id", "name", "city"])

print("📥 Incoming Source Records:")
source_df.show(truncate=False)

Expected Behaviour:

Step 4: Detect Changes by Comparing Source vs. Dimension

The core logic of SCD-2 is detecting which records actually changed. We join the source with the current dimension snapshot and compare business attributes:

# Load the current dimension table
dim_table = DeltaTable.forPath(spark, DIMENSION_PATH)
dim_current = dim_table.toDF().filter(F.col("is_current") == True)

# Join source with current dimension on natural key
comparison_df = source_df.alias("src").join(
    dim_current.alias("dim"),
    on=F.col("src.customer_id") == F.col("dim.customer_id"),
    how="left"
)

# Classify each record
classified_df = comparison_df.withColumn(
    "action",
    F.when(F.col("dim.customer_id").isNull(), F.lit("INSERT"))        # New customer
     .when(
         (F.col("src.name") != F.col("dim.name")) |                   # Name changed
         (F.col("src.city") != F.col("dim.city")),                    # City changed
         F.lit("UPDATE")
     )
     .otherwise(F.lit("RETAIN"))                                       # No change
)

print("🔍 Change Classification:")
classified_df.select("src.customer_id", "src.city", "dim.city", "action").show(truncate=False)

Sample Classification Output:

+-----------+---------+------+--------+
|customer_id|src_city |dim_city|action |
+-----------+---------+------+--------+
|C101       |Bangalore|Mumbai |UPDATE  |
|C103       |Delhi    |Delhi  |RETAIN  |
|C104       |Hyderabad|null   |INSERT  |
+-----------+---------+------+--------+

Step 5: Build the SCD-2 Merge Logic

This is the most critical step. For SCD-2, we need to perform two simultaneous operations in a single atomic transaction:

1. Expire the old row: Set is_current = false and end_date = today on the existing active record.
2. Insert the new version: Add a new row with updated attributes, is_current = true, start_date = today, and end_date = 9999-12-31.

Pure PySpark Implementation (Without Delta MERGE):

from pyspark.sql.window import Window
from datetime import date

today = date.today()
FAR_FUTURE = date(9999, 12, 31)

# -------------------------------------------------------
# 1. Get the max surrogate key for generating new IDs
# -------------------------------------------------------
max_sk = dim_table.toDF().agg(F.max("surrogate_key")).collect()[0][0] or 0

# -------------------------------------------------------
# 2. UNCHANGED rows — Keep exactly as-is
# -------------------------------------------------------
unchanged_df = classified_df.filter(F.col("action") == "RETAIN") \
    .select("dim.*")

# -------------------------------------------------------
# 3. EXPIRED rows — Close out the old version of changed records
# -------------------------------------------------------
expired_df = classified_df.filter(F.col("action") == "UPDATE") \
    .select("dim.*") \
    .withColumn("end_date", F.lit(today)) \
    .withColumn("is_current", F.lit(False))

# -------------------------------------------------------
# 4. NEW VERSION rows — Insert updated attributes as new current records
# -------------------------------------------------------
# Generate monotonically increasing surrogate keys
new_version_df = classified_df.filter(F.col("action") == "UPDATE") \
    .select(
        F.col("src.customer_id"),
        F.col("src.name"),
        F.col("src.city"),
    ) \
    .withColumn("start_date", F.lit(today)) \
    .withColumn("end_date", F.lit(FAR_FUTURE)) \
    .withColumn("is_current", F.lit(True)) \
    .withColumn("surrogate_key", F.monotonically_increasing_id() + max_sk + 1)

# -------------------------------------------------------
# 5. BRAND NEW rows — Insert customers never seen before
# -------------------------------------------------------
brand_new_df = classified_df.filter(F.col("action") == "INSERT") \
    .select(
        F.col("src.customer_id"),
        F.col("src.name"),
        F.col("src.city"),
    ) \
    .withColumn("start_date", F.lit(today)) \
    .withColumn("end_date", F.lit(FAR_FUTURE)) \
    .withColumn("is_current", F.lit(True)) \
    .withColumn("surrogate_key", F.monotonically_increasing_id() + max_sk + 100)

# -------------------------------------------------------
# 6. Also include dimension records NOT in today's source (untouched)
# -------------------------------------------------------
untouched_df = dim_table.toDF().alias("dim").join(
    source_df.alias("src"),
    on=F.col("dim.customer_id") == F.col("src.customer_id"),
    how="left_anti"
)

# -------------------------------------------------------
# 7. UNION ALL components into the final dimension snapshot
# -------------------------------------------------------
# Align all DataFrames to the same column order
column_order = ["surrogate_key", "customer_id", "name", "city", "start_date", "end_date", "is_current"]

final_dim_df = unchanged_df.select(column_order) \
    .unionByName(expired_df.select(column_order)) \
    .unionByName(new_version_df.select(column_order)) \
    .unionByName(brand_new_df.select(column_order)) \
    .unionByName(untouched_df.select(column_order))

print("✅ Final SCD-2 Dimension Table After Merge:")
final_dim_df.orderBy("customer_id", "start_date").show(truncate=False)

Expected Final Output:

+-------------+-------------+------------+-----------+------------+------------+----------+
|surrogate_key|customer_id  |name        |city       |start_date  |end_date    |is_current|
+-------------+-------------+------------+-----------+------------+------------+----------+
|1            |C101         |Amit Kumar  |Mumbai     |2024-01-15  |2026-05-31  |false     |  ← Expired old row
|4            |C101         |Amit Kumar  |Bangalore  |2026-05-31  |9999-12-31  |true      |  ← New current version
|2            |C102         |Priya Rao   |Pune       |2024-03-01  |9999-12-31  |true      |  ← Untouched (not in source)
|3            |C103         |Mukesh Sen  |Delhi      |2024-06-10  |9999-12-31  |true      |  ← Retained (no change)
|5            |C104         |Neha Gupta  |Hyderabad  |2026-05-31  |9999-12-31  |true      |  ← Brand new insert
+-------------+-------------+------------+-----------+------------+------------+----------+

Notice how C101 now has two rows — the old historical version (Mumbai, expired) and the new current version (Bangalore, active). This is exactly what SCD-2 achieves: full attribute-level history tracking.

Step 6: Delta Lake MERGE Alternative (Production Recommended)

In production Delta Lake / Databricks environments, the MERGE INTO command provides an atomic, ACID-compliant way to perform the SCD-2 logic in a single statement:

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

today = date.today()
FAR_FUTURE = date(9999, 12, 31)

dim_table = DeltaTable.forPath(spark, DIMENSION_PATH)

# Prepare staged updates: records that actually changed
staged_updates = source_df.alias("src").join(
    dim_table.toDF().filter("is_current = true").alias("dim"),
    "customer_id"
).where("src.name != dim.name OR src.city != dim.city") \
 .select("src.*")

# MERGE operation
dim_table.alias("target").merge(
    staged_updates.alias("source"),
    "target.customer_id = source.customer_id AND target.is_current = true"
).whenMatchedUpdate(
    set={
        "is_current": F.lit(False),
        "end_date": F.lit(today)
    }
).whenNotMatchedInsert(
    values={
        "surrogate_key": F.monotonically_increasing_id(),
        "customer_id": F.col("source.customer_id"),
        "name": F.col("source.name"),
        "city": F.col("source.city"),
        "start_date": F.lit(today),
        "end_date": F.lit(FAR_FUTURE),
        "is_current": F.lit(True)
    }
).execute()

# Separately insert new version rows for matched (changed) records
new_rows = staged_updates.withColumn("surrogate_key", F.monotonically_increasing_id() + 1000) \
    .withColumn("start_date", F.lit(today)) \
    .withColumn("end_date", F.lit(FAR_FUTURE)) \
    .withColumn("is_current", F.lit(True))

new_rows.write.format("delta").mode("append").save(DIMENSION_PATH)

# Also handle brand new customers
new_customers = source_df.alias("src").join(
    dim_table.toDF().alias("dim"),
    on="customer_id",
    how="left_anti"
).withColumn("surrogate_key", F.monotonically_increasing_id() + 2000) \
 .withColumn("start_date", F.lit(today)) \
 .withColumn("end_date", F.lit(FAR_FUTURE)) \
 .withColumn("is_current", F.lit(True))

new_customers.write.format("delta").mode("append").save(DIMENSION_PATH)

print("✅ Delta MERGE SCD-2 completed successfully.")

Follow-Up Questions & Answers

Q1: Why use SCD-2 instead of SCD-1?

A: SCD-1 simply overwrites the old attribute value with the new one. You lose all historical context. For example, if a customer moves from Mumbai to Bangalore, SCD-1 would update the city column in-place — you'd never know the customer was previously in Mumbai. SCD-2 preserves every historical version, enabling time-travel queries like: "Which city was this customer in on 2024-06-15?"

Q2: How do you query the dimension for a specific historical point in time?

A: Use the start_date and end_date range to filter:

# Get the customer snapshot as it existed on 2024-08-01
point_in_time = "2024-08-01"

historical_df = dim_df.filter(
    (F.col("start_date") <= point_in_time) &
    (F.col("end_date") > point_in_time)
)

historical_df.show(truncate=False)

This returns exactly one row per customer_id — the version that was active on that specific date.

Q3: How do you handle late-arriving data in SCD-2?

A: Late-arriving data is a record that should have been processed in a prior batch but arrives in a current batch. Handling this requires:

1. Insert the late record with the correct start_date (when the change actually occurred, not today).
2. Split the existing historical range if the late record falls inside an already-closed time range.
3. Re-sequence the end_dates of surrounding records.

This is significantly more complex and often handled by custom logic or specialized frameworks like Apache Hudi's timeline-based upserts.

Q4: What is the difference between is_current flag and end_date = 9999-12-31?

A: Both serve the same logical purpose (identifying the active row), but they are used differently:

• is_current flag: Simple boolean filter. Fast for WHERE is_current = true queries. Optimized by predicate pushdown.
• end_date = 9999-12-31: Enables time-range queries (WHERE start_date <= '2024-08-01' AND end_date > '2024-08-01'). Essential for point-in-time reporting.

Best practice is to maintain both columns for maximum query flexibility.

Q5: How does SCD-2 handle DELETE operations?

A: SCD-2 traditionally does not physically delete rows. Instead, you perform a soft delete:

• Set is_current = false and end_date = today on the active row.
• Optionally, add a is_deleted flag set to true to distinguish deletions from updates.

# Soft delete: expire the record
expired = dim_df.filter(
    (F.col("customer_id") == "C102") & (F.col("is_current") == True)
).withColumn("is_current", F.lit(False)) \
 .withColumn("end_date", F.lit(today))

Sub-Scenarios

Sub-Scenario A: SCD-2 with Multiple Attribute Changes in Same Batch

Situation: The source feed contains two records for the same customer in a single batch (e.g., customer moved cities twice in one day).

Fix: Order the source records by a timestamp column and apply a row_number() window to process them sequentially:

from pyspark.sql.window import Window

window = Window.partitionBy("customer_id").orderBy(F.col("event_timestamp").asc())

# Keep only the latest change per customer per batch
deduplicated_source = source_df.withColumn("rn", F.row_number().over(window)) \
    .filter(F.col("rn") == F.count("*").over(Window.partitionBy("customer_id")))

Sub-Scenario B: SCD-2 at Massive Scale (Billions of Rows)

Situation: Your dimension table has 2 billion rows and the daily merge is extremely slow.

Fix:

1. Partition the Delta table by is_current so that active rows are in a small, fast-scanning partition: python dim_df.write.format("delta").partitionBy("is_current").save(DIMENSION_PATH)

2. Z-ORDER by customer_id to co-locate all versions of the same customer physically on disk: sql OPTIMIZE dim_customer ZORDER BY (customer_id)

3. Use Delta's MERGE which leverages data skipping and file-level statistics to avoid full table scans.

Sub-Scenario C: SCD-2 Without Delta Lake (Plain Parquet)

Situation: Your environment doesn't support Delta Lake, Iceberg, or Hudi. You're working with raw Parquet on S3.

Fix: Use the full-snapshot rebuild approach (Step 5 above):

1. Read the entire existing dimension.
2. Classify records (INSERT / UPDATE / RETAIN).
3. Build the new snapshot via unionByName.
4. Overwrite the entire dimension table: python final_dim_df.write.mode("overwrite").parquet(DIMENSION_PATH)

⚠️ Warning: This is significantly more expensive than Delta MERGE because you rewrite all historical rows every batch. For large dimensions, migrate to Delta Lake or Iceberg.