Upserts and slowly changing dimensions, explained simply

Hook: Why your customer data keeps getting out of sync

You've just deployed a new feature that flags customers as premium after their first purchase. A week later, you notice some accounts are showing both a free and premium status in your reports. The problem isn't your application—it's how you're handling data updates. When we talk about upserts and slowly changing dimensions, explained simply, we're solving exactly this: how to manage updates to dimension tables while keeping a complete audit trail. This is one of the most common data engineering challenges in production systems, and getting it right means the difference between reliable analytics and spreadsheet-driven firefighting.

What are slowly changing dimensions?

A slowly changing dimension (SCD) is a dimension table where attributes change over time, but not frequently. Think of a customer table: most of the time a customer's email stays the same, but occasionally they update their address, their tier changes, or their status gets corrected. The question is: how do you record these changes?

Without a strategy, you either lose history (if you just overwrite) or end up with duplicate records and confusion. Data warehouses need to handle this deliberately because your business stakeholders will ask questions like "how many customers were premium on March 15?" or "when did this account's status actually change?"

Slowly changing dimensions come in several types, but the two you'll encounter in 95% of production systems are Type 1 and Type 2. Each answers a different business question and comes with trade-offs around storage, query complexity, and historical accuracy.

SCD Type 1: Overwrite values

Type 1 is the simplest approach: when a dimension attribute changes, you just replace the old value. There is no history kept—only the current state exists.

When to use Type 1: When you only care about current state and historical changes don't matter. Examples include correcting data quality issues (a misspelled city name), or refreshing data that's known to be constantly updated and irrelevant in the past (like a last-login timestamp or an internal scoring field).

Here's a concrete example. You have a products table and prices change frequently, but you never need to report "what was the price on February 10?" You just need today's price for margin calculations.

from pyspark.sql import SparkSession
from pyspark.sql.functions import current_date, lit

spark = SparkSession.builder.appName("SCD1Example").getOrCreate()

# Existing product dimension
products = spark.createDataFrame([
    (1, "Laptop", 999.99),
    (2, "Monitor", 299.99),
    (3, "Keyboard", 79.99),
], ["product_id", "product_name", "price"])

# New price updates (late-arriving data or corrections)
price_updates = spark.createDataFrame([
    (2, "Monitor", 279.99),  # Price dropped
    (3, "Keyboard", 89.99),  # Price increased
], ["product_id", "product_name", "price"])

# Type 1: Simple merge—just overwrite
products.createOrReplaceTempView("products")
price_updates.createOrReplaceTempView("price_updates")

result = spark.sql("""
    MERGE INTO products p
    USING price_updates u
    ON p.product_id = u.product_id
    WHEN MATCHED THEN
        UPDATE SET p.price = u.price
    WHEN NOT MATCHED THEN
        INSERT (product_id, product_name, price) 
        VALUES (u.product_id, u.product_name, u.price)
""")

# Result: products now has the new prices. Old prices are gone.

Type 1 is fast and storage-efficient, but it's lossy. You can't answer historical questions. Use it when the business has decided that historical tracking isn't required.

SCD Type 2: Keep history with validity dates

Type 2 tracks every change by adding temporal columns. Each version of a record exists as a separate row, stamped with validity dates (effective date and end date). When an attribute changes, you close the old record and open a new one.

When to use Type 2: When you need to answer "what did this look like on that date?" This is essential for compliance, accurate historical reporting, and auditing. Most customer, employee, and account dimensions use Type 2.

Real-world scenario: A customer's tier changes from Standard to Premium on June 15. With Type 2, you keep both records—the Standard record is marked valid until June 14, and a new Premium record starts June 15. When you run a revenue report as-of June 10, you use only the Standard record. When you run it as-of June 20, you use the Premium record.

from pyspark.sql import SparkSession
from pyspark.sql.functions import current_date, lit, col, max as spark_max, when

spark = SparkSession.builder.appName("SCD2Example").getOrCreate()

# Existing customer dimension with validity dates
customers = spark.createDataFrame([
    (1, "Alice", "Standard", "2024-01-01", "2099-12-31", True),
    (2, "Bob", "Premium", "2024-03-15", "2099-12-31", True),
    (3, "Charlie", "Standard", "2024-02-01", "2099-12-31", True),
], ["customer_id", "customer_name", "tier", "effective_date", "end_date", "is_current"])

# Updates: Alice and Charlie tier changes
tier_updates = spark.createDataFrame([
    (1, "Alice", "Premium", "2024-06-15"),  # Late-arriving data: Alice upgraded
    (3, "Charlie", "Gold", "2024-06-10"),   # Charlie promoted early
], ["customer_id", "customer_name", "tier", "effective_date"])

customers.createOrReplaceTempView("customers")
tier_updates.createOrReplaceTempView("tier_updates")

# Type 2: Insert new records and close old ones
result = spark.sql("""
    MERGE INTO customers c
    USING tier_updates u
    ON c.customer_id = u.customer_id AND c.is_current = True
    WHEN MATCHED THEN
        UPDATE SET 
            c.end_date = DATE_SUB(u.effective_date, 1),
            c.is_current = False
    WHEN NOT MATCHED THEN
        INSERT (customer_id, customer_name, tier, effective_date, end_date, is_current)
        VALUES (u.customer_id, u.customer_name, u.tier, u.effective_date, '2099-12-31', True)
""")

# After merge:
# (1, "Alice", "Standard", "2024-01-01", "2024-06-14", False)  -- closed
# (1, "Alice", "Premium", "2024-06-15", "2099-12-31", True)    -- new version
# (2, "Bob", "Premium", "2024-03-15", "2099-12-31", True)      -- unchanged
# (3, "Charlie", "Standard", "2024-02-01", "2024-06-09", False) -- closed
# (3, "Charlie", "Gold", "2024-06-10", "2099-12-31", True)     -- new version

Type 2 preserves complete history but requires more storage and careful query logic. You must always filter on validity dates or the is_current flag to avoid double-counting in reports.

MERGE statements for both approaches

Apache Spark's MERGE command (available in Delta Lake and other engines supporting ACID transactions) makes both SCD implementations straightforward. The key is understanding the difference in the UPDATE clause.

For Type 1: MERGE finds matching records and updates the column directly. It's a one-step operation.

For Type 2: MERGE finds the current record, closes it (sets end_date and is_current), then the NOT MATCHED clause inserts the new version. You're effectively doing two operations in one transaction.

Here's a production-ready Type 2 pattern that handles the edge case where multiple updates arrive in the same batch:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, row_number, max as spark_max
from pyspark.sql.window import Window

spark = SparkSession.builder.appName("RobustSCD2").getOrCreate()

# Dimension table
spark.sql("""
    CREATE TABLE IF NOT EXISTS gold.customers (
        customer_id INT,
        customer_name STRING,
        tier STRING,
        effective_date DATE,
        end_date DATE,
        is_current BOOLEAN
    )
    USING DELTA
""")

# Updates with potential duplicates
updates_raw = spark.createDataFrame([
    (1, "Alice", "Premium", "2024-06-15"),
    (1, "Alice", "Gold", "2024-06-20"),     # Alice upgraded again—keep newest
    (2, "Bob", "Premium", "2024-06-18"),
], ["customer_id", "customer_name", "tier", "effective_date"])

# Deduplicate: keep only the latest change per customer if multiple in batch
window_spec = Window.partitionBy("customer_id").orderBy(col("effective_date").desc())
tier_updates = updates_raw.withColumn("rn", row_number().over(window_spec)) \
    .filter(col("rn") == 1) \
    .drop("rn")

tier_updates.createOrReplaceTempView("tier_updates")

# Type 2 merge with de-duplication
spark.sql("""
    MERGE INTO gold.customers c
    USING tier_updates u
    ON c.customer_id = u.customer_id AND c.is_current = True
    WHEN MATCHED THEN
        UPDATE SET 
            c.end_date = DATE_SUB(u.effective_date, 1),
            c.is_current = False
    WHEN NOT MATCHED THEN
        INSERT (customer_id, customer_name, tier, effective_date, end_date, is_current)
        VALUES (u.customer_id, u.customer_name, u.tier, u.effective_date, '2099-12-31', True)
""")

The MERGE statement is transactional, so either all updates succeed or none do. This prevents the inconsistent state where an old record is closed but the new one isn't inserted.

Test with late-arriving data

In real systems, data doesn't always arrive in order. A transaction from June 10 might land in your warehouse on June 25. If your SCD logic doesn't handle this, you'll backfill changes incorrectly and corrupt your audit trail.

Late-arriving data scenario: On June 1, your system records that Alice was Standard tier. On June 15, you receive an update from a source system saying Alice upgraded to Premium on June 10 (it just took a week to propagate). Your warehouse should insert the Premium record with effective_date of June 10, not June 15.

Here's how to test this:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit
from datetime import datetime, timedelta

spark = SparkSession.builder.appName("LateArrivingTest").getOrCreate()

# Initial state: Alice as Standard
spark.createDataFrame([
    (1, "Alice", "Standard", "2024-06-01", "2099-12-31", True),
], ["customer_id", "customer_name", "tier", "effective_date", "end_date", "is_current"]) \
    .write.mode("overwrite").saveAsTable("customers_test")

# Simulate: it's now June 25, and we receive a late-arriving update
# (Alice upgraded to Premium on June 10, but we're just finding out)
late_update = spark.createDataFrame([
    (1, "Alice", "Premium", "2024-06-10"),
], ["customer_id", "customer_name", "tier", "effective_date"])

late_update.createOrReplaceTempView("late_update")

spark.sql("""
    MERGE INTO customers_test c
    USING late_update u
    ON c.customer_id = u.customer_id AND c.is_current = True
    WHEN MATCHED THEN
        UPDATE SET 
            c.end_date = DATE_SUB(u.effective_date, 1),
            c.is_current = False
    WHEN NOT MATCHED THEN
        INSERT (customer_id, customer_name, tier, effective_date, end_date, is_current)
        VALUES (u.customer_id, u.customer_name, u.tier, u.effective_date, '2099-12-31', True)
""")

# Result: 
# (1, "Alice", "Standard", "2024-06-01", "2024-06-09", False)  -- closed, valid until June 9
# (1, "Alice", "Premium", "2024-06-10", "2099-12-31", True)    -- new version, effective June 10

# Test query: "What was Alice's tier on June 5?" Should return Standard
result = spark.sql("""
    SELECT tier 
    FROM customers_test 
    WHERE customer_id = 1 
    AND effective_date <= '2024-06-05' 
    AND end_date >= '2024-06-05'
""")
result.show()  # Output: Standard ✓

# Test query: "What is Alice's tier today?" Should return Premium
result = spark.sql("""
    SELECT tier 
    FROM customers_test 
    WHERE customer_id = 1 
    AND is_current = True
""")
result.show()  # Output: Premium ✓

Notice how the late-arriving update correctly backdates the Premium record to June 10 and closes the Standard record at June 9. Your historical query as-of June 5 returns the right answer. This is why effective_date in the MERGE condition (not "today's date") is critical.

Common mistakes

• Using current_date() instead of the source effective_date: When you set end_date = current_date() on an old record, you lose the accurate point-in-time when the change occurred. Always use the effective_date from the update source.
• Forgetting the is_current flag: Type 2 tables without an is_current column force every query to join on date ranges. A boolean flag makes filters fast and prevents accidental double-counting.
• Not deduplicating within a batch: If two updates for the same customer arrive in the same load, your MERGE can fail or produce duplicate records. Deduplicate before merging.
• Mixing Type 1 and Type 2 in the same table: Some columns change frequently (Type 1) and others rarely (Type 2). Use separate columns or separate tables for each SCD type to avoid confusion. Don't try to handle both in one MERGE statement.
• Assuming MERGE is idempotent: If you run the same MERGE twice without deduplication, you'll insert duplicate records. Add idempotency checks if your data pipeline retries.

Key takeaways

• SCD Type 1 (overwrite) is appropriate only when you genuinely don't need history. Use it for corrections and metrics that are never queried historically.
• SCD Type 2 (validity dates) is the standard for customer, product, and account dimensions. It preserves audit trails and enables point-in-time reporting.
• MERGE statements handle both SCD types, but the logic differs: Type 1 updates a column directly; Type 2 closes old records and inserts new ones in a single transaction.
• Late-arriving data is inevitable. Use the source's effective_date in your MERGE condition, not the processing date, to keep history accurate.
• Always include an is_current flag or end_date range in your dimension queries to avoid counting obsolete records and producing inflated metrics.