Implementing Slowly Changing Dimension Type-2 (SCD Type-2) in PySpark with Delta Lake

Introduction

Slowly Changing Dimensions (SCD) are critical in data warehousing to track changes in dimension tables over time. SCD Type-2 retains historical data by inserting new records for updates while marking old records as inactive. This method helps in maintaining data integrity and ensuring accurate historical tracking.

In this article, we will implement SCD Type-2 using PySpark and Delta Lake.

Why Use SCD Type-2?

SCD Type-2 is essential when:

• We need to maintain historical records of changes in data.
• Analytics require tracking attribute changes over time.
• Business reporting depends on historical insights rather than just current values.

Now, let’s dive into the implementation.

Implementation in PySpark

We will create a Delta table, insert initial records, and then perform an SCD Type-2 update.

Step 1: Import Libraries

from delta.tables import DeltaTable
from pyspark.sql.types import StringType, IntegerType, StructType, StructField
from pyspark.sql.functions import col, lit, current_timestamp

Step 2: Create Delta Table

# Define a Delta Table with necessary columns
DeltaTable.createIfNotExists(spark)\
    .tableName("employees")\
    .addColumn("id", "INT")\
    .addColumn("name", "STRING")\
    .addColumn("age", "INT")\
    .addColumn("salary", "INT")\
    .addColumn("is_active", "STRING")\
    .addColumn("created_at", "TIMESTAMP")\
    .addColumn("updated_at", "TIMESTAMP")\
    .property("description", "employee table")\
    .location("/FileStore/tables/delta/employees")\
    .execute()

Step 3: Insert Initial Data

We insert two employee records with an is_active status as Y (active).

# Define schema for the DataFrame
schema = StructType([
    StructField("id", IntegerType()),
    StructField("name", StringType()),
    StructField("age", IntegerType()),
    StructField("salary", IntegerType())
])

# Sample data for initial insert
initial_data = [(1, "John", 32, 30000), (2, "Jack", 32, 30000)]

# Create DataFrame
df = spark.createDataFrame(data=initial_data, schema=schema)

# Add is_active and timestamp columns
final_df = df.withColumn("is_active", lit("Y"))\
             .withColumn("created_at", current_timestamp())\
             .withColumn("updated_at", current_timestamp())

# Insert into Delta table
final_df.write.mode("append").option("mergeSchema", True).saveAsTable("employees")

Step 4: Display Initial Data

# Read the table to verify inserted records
employees_table = DeltaTable.forPath(spark, "/FileStore/tables/delta/employees")
display(employees_table.toDF())

Step 5: Perform SCD Type-2 Update

We assume that employee Jack (id = 2) gets a salary raise and his age is updated. Instead of modifying the existing record, we:

1. Mark the old record as inactive (is_active='N').
2. Insert a new record with the updated details.

# New data representing an update for employee with id = 2
update_data = [(2, "Jack", 36, 70000)]
df_update = spark.createDataFrame(data=update_data, schema=schema)

# Load Delta table reference
employees_table = DeltaTable.forPath(spark, "/FileStore/tables/delta/employees")

# Update condition based on matching id
condition = "original.id = new.id and original.is_active = 'Y'"

# Mark the existing record as inactive
employees_table.alias("original").merge(
    df_update.alias("new"), condition
).whenMatchedUpdate(
    set={
        "is_active": lit("N"),
        "updated_at": current_timestamp()
    }
).execute()

# Insert the updated record as a new entry with is_active = 'Y'
new_df = df_update.withColumn("is_active", lit("Y"))\
                  .withColumn("created_at", current_timestamp())\
                  .withColumn("updated_at", current_timestamp())

new_df.write.mode("append").option("mergeSchema", True).saveAsTable("employees")

Step 6: Verify the Update

# Display updated table
employees_table = DeltaTable.forPath(spark, "/FileStore/tables/delta/employees")
display(employees_table.toDF())

Expected Results

id name age salary is_active created_at updated_at

1 John 32 30000 Y … …

2 Jack 32 30000 N … …

2 Jack 36 70000 Y … …

Conclusion

SCD Type-2 is essential for tracking historical changes in data. By implementing this in PySpark with Delta Lake, we:

• Maintain historical records of changes.
• Ensure that no data is lost while updating.
• Enable better analytical insights based on past trends.

This method is widely used in data warehousing and ETL processes where tracking historical changes is crucial for decision-making. With PySpark and Delta Lake, we can efficiently implement SCD Type-2 while benefiting from Delta Lake’s transactional consistency and performance optimizations.