SCD Type 2 Done Right: Declarative Pipelines and Change Data Feed

Every data warehouse eventually hits the same wall: someone updates a customer record, and the old value vanishes. The analyst running a report on last quarter’s regional breakdown gets this quarter’s data instead. The audit team asks “what was this customer’s tier on March 15th?” and nobody can answer.

SCD Type 2 solves this by keeping every version of a record with validity timestamps. The problem isn’t understanding the concept. The problem is implementing it without drowning in merge logic, deduplication edge cases, and late-arriving deletes.

Databricks Declarative Pipelines (what used to be called Delta Live Tables) has a feature called APPLY CHANGES that handles all of this. You declare what the keys are, what column to sequence by, and which columns to track. The framework generates the merge logic, handles out-of-order events, and maintains the __START_AT / __END_AT timestamps for you.

What SCD Type 2 Actually Looks Like

#

A standard dimension table stores the current state:

An SCD Type 2 table stores every state:

The row with __END_AT = NULL is the current version. Historical queries filter by date range. Simple in theory. In practice, you need to handle:

• Out-of-order arrivals: An update with updated_at = 2025-05-15 arrives after one with updated_at = 2025-06-01
• Duplicate events: The same update is delivered twice by the CDC source
• Deletes: The source system removes a record, and you need to close out the SCD chain
• Late-arriving deletes: A delete event arrives after newer updates have been processed

Writing this by hand means a multi-way MERGE statement with careful timestamp arithmetic. Every team that does this builds slightly different logic, and the edge cases bite at 3am.

Architecture

#

flowchart LR
    subgraph Source["CDC Source"]
        S["Cloud Storage
(JSON/Parquet CDC events)"]
    end

    subgraph Pipeline["Declarative Pipeline"]
        R["customers_raw
(streaming table)"]
        SCD["customers_scd2
(APPLY CHANGES)"]
    end

    subgraph Downstream["Consumers"]
        CDF["Change Data Feed"]
        D["Dashboard / Analytics"]
    end

    S --> R --> SCD
    SCD --> CDF --> D

    style S fill:#3498db,color:white
    style R fill:#e67e22,color:white
    style SCD fill:#2ecc71,color:white
    style CDF fill:#9b59b6,color:white
    style D fill:#f39c12,color:white

Step 1: Set Up the Landing Zone

#

Create a catalog, schema, and volume for the CDC data:

1
2
3

CREATE CATALOG IF NOT EXISTS retail_demo;
CREATE SCHEMA IF NOT EXISTS retail_demo.customers;
CREATE VOLUME IF NOT EXISTS retail_demo.customers.cdc_landing;

Generate synthetic CDC events. In production, these come from Debezium, Fivetran, or a similar tool:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

import json
import random
from datetime import datetime, timedelta, timezone

CUSTOMERS = [f"C{i:03d}" for i in range(1, 51)]
TIERS = ["bronze", "silver", "gold", "platinum"]
REGIONS = ["US-WEST", "US-EAST", "EU-WEST", "EU-EAST", "APAC"]
OPERATIONS = ["INSERT", "UPDATE", "UPDATE", "UPDATE", "DELETE"]

def generate_cdc_events(n_events=200):
    events = []
    base_time = datetime(2025, 1, 1, tzinfo=timezone.utc)

    for i in range(n_events):
        customer_id = random.choice(CUSTOMERS)
        op = random.choice(OPERATIONS)
        ts = base_time + timedelta(hours=random.randint(0, 8760))

        event = {
            "customer_id": customer_id,
            "operation": op,
            "updated_at": ts.isoformat(),
        }

        if op != "DELETE":
            event.update({
                "email": f"{customer_id.lower()}@example.com",
                "full_name": f"Customer {customer_id}",
                "tier": random.choice(TIERS),
                "region": random.choice(REGIONS),
                "lifetime_value": round(random.uniform(100, 50000), 2),
            })

        events.append(event)

    return events

# Write events as newline-delimited JSON
volume_path = "/Volumes/retail_demo/customers/cdc_landing"
events = generate_cdc_events(200)

# Split into 4 batches to simulate incremental arrival
batch_size = 50
for i in range(0, len(events), batch_size):
    batch = events[i:i + batch_size]
    file_path = f"{volume_path}/cdc_batch_{i // batch_size:04d}.json"
    with open(file_path, "w") as f:
        for e in batch:
            f.write(json.dumps(e) + "\n")

print(f"Wrote {len(events)} CDC events in 4 batches")

Step 2: Build the Declarative Pipeline

#

Create a new Declarative Pipeline (DLT) notebook. The pipeline has two tables: a raw ingestion table and the SCD2 target.

Raw Ingestion

#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

import dlt
from pyspark.sql import functions as F

@dlt.table(
    comment="Raw CDC events from customer source system"
)
def customers_raw():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .load("/Volumes/retail_demo/customers/cdc_landing/")
        .withColumn("_ingested_at", F.current_timestamp())
    )

SCD Type 2 with APPLY CHANGES

#

This is where the magic happens. Instead of writing merge logic, you declare the transformation:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

dlt.create_streaming_table(
    name="customers_scd2",
    comment="SCD Type 2 customer dimension with full history"
)

dlt.apply_changes(
    target="customers_scd2",
    source="customers_raw",
    keys=["customer_id"],
    sequence_by="updated_at",
    apply_as_deletes=F.expr("operation = 'DELETE'"),
    apply_as_truncates=None,
    column_list=[
        "customer_id",
        "email",
        "full_name",
        "tier",
        "region",
        "lifetime_value",
    ],
    track_history_column_list=[
        "email", "tier", "region", "lifetime_value"
    ],
    stored_as_scd_type=2,
)

What APPLY CHANGES does under the hood:

1. Sequences by updated_at – so out-of-order events get inserted at the correct position in the history chain
2. Tracks changes on specified columns – only creates a new SCD2 row when email, tier, region, or lifetime_value actually changes
3. Handles deletes – when operation = 'DELETE', it closes the current SCD2 row by setting __END_AT
4. Deduplicates – if the same event arrives twice with the same updated_at, it’s idempotent

Step 3: Deploy and Run

#

Configure the pipeline in the Databricks UI or via the API:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

{
  "name": "customer_scd2_pipeline",
  "catalog": "retail_demo",
  "schema": "customers",
  "libraries": [
    {"notebook": {"path": "/Workspace/Users/you/customer_scd2_notebook"}}
  ],
  "continuous": false,
  "development": true,
  "channel": "CURRENT"
}

Run the pipeline. On first execution, it processes all existing files. Subsequent runs pick up only new files.

Step 4: Query the SCD2 Table

#

Current State (Latest Version)

#

1
2
3
4
5

-- Current customers: __END_AT is NULL
SELECT customer_id, email, tier, region, lifetime_value, __START_AT
FROM retail_demo.customers.customers_scd2
WHERE __END_AT IS NULL
ORDER BY customer_id

Point-in-Time Query

#

1
2
3
4
5
6
7

-- What was the customer state on June 15, 2025?
SELECT customer_id, email, tier, region, lifetime_value,
       __START_AT, __END_AT
FROM retail_demo.customers.customers_scd2
WHERE __START_AT <= '2025-06-15'
  AND (__END_AT > '2025-06-15' OR __END_AT IS NULL)
ORDER BY customer_id

Who Changed Tier Recently?

#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

-- Tier changes in the last 30 days
SELECT
    customer_id,
    tier AS current_tier,
    __START_AT AS changed_at,
    LAG(tier) OVER (
        PARTITION BY customer_id ORDER BY __START_AT
    ) AS previous_tier
FROM retail_demo.customers.customers_scd2
WHERE __START_AT >= current_date() - 30
ORDER BY __START_AT DESC

Step 5: Enable Change Data Feed for Downstream

#

Change Data Feed (CDF) lets downstream consumers read only the rows that changed, instead of rescanning the full table:

1
2
3

-- CDF is already enabled by Declarative Pipelines on SCD2 tables
-- Verify:
DESCRIBE TABLE EXTENDED retail_demo.customers.customers_scd2

Read the changes incrementally:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# Read CDF as a stream for downstream processing
cdf_stream = (
    spark.readStream
    .format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 0)
    .table("retail_demo.customers.customers_scd2")
)

# Each row includes _change_type: insert, update_preimage,
# update_postimage, delete
display(cdf_stream)

1
2
3
4

-- Batch CDF read: what changed between versions 2 and 5?
SELECT customer_id, tier, region, _change_type, _commit_version
FROM table_changes('retail_demo.customers.customers_scd2', 2, 5)
ORDER BY _commit_version, customer_id

Testing the Edge Cases

#

The point of using APPLY CHANGES is that it handles the hard cases. Here’s how to verify:

Out-of-Order Events

#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# Write an event with an old timestamp after newer ones exist
late_event = [{
    "customer_id": "C001",
    "operation": "UPDATE",
    "updated_at": "2025-03-15T00:00:00+00:00",
    "email": "[email protected]",
    "full_name": "Customer C001",
    "tier": "silver",
    "region": "EU-WEST",
    "lifetime_value": 5000.00,
}]

# Write to landing zone and trigger pipeline
import json
with open(f"{volume_path}/late_event.json", "w") as f:
    for e in late_event:
        f.write(json.dumps(e) + "\n")

After the pipeline runs, check that the SCD2 chain for C001 correctly inserts the silver/EU-WEST row at its proper chronological position, splitting existing ranges if needed.

Duplicate Events

#

Write the same event twice with identical updated_at. The pipeline should produce the same result as processing it once. Run the pipeline, check row counts – no duplicates.

Late Delete

#

1
2
3
4
5
6

# Delete event arrives with a timestamp between two existing updates
delete_event = [{
    "customer_id": "C001",
    "operation": "DELETE",
    "updated_at": "2025-04-01T00:00:00+00:00",
}]

The pipeline should close the SCD2 row that was active at 2025-04-01 without affecting rows with later __START_AT values.

Gotchas

#

__START_AT / __END_AT semantics. These are generated and managed by the framework. Don’t try to set them manually. __START_AT is inclusive, __END_AT is exclusive. A row with __END_AT = NULL is the current version.

Pipeline retry behavior. If a pipeline run fails mid-batch, the next run replays from the last checkpoint. Because APPLY CHANGES uses sequence-based deduplication, replayed events don’t create duplicate SCD2 rows. But this means your CDC source must be replayable (which cloud storage naturally is, Kafka with retention also works).

Column list matters. If you omit a column from track_history_column_list, changes to that column won’t trigger a new SCD2 row. This is a feature, not a bug – you probably don’t want a new row every time last_login_at updates. Be intentional about what you track.

Late-arriving deletes. When a delete event arrives with a timestamp that falls between existing SCD2 rows, the framework handles it correctly by adjusting the chain. But if deletes arrive very late (weeks after the fact), you should verify the resulting history makes business sense. The framework is mechanically correct, but “customer was deleted on March 1st” arriving in April might need manual review.

Handling schema evolution. If the source adds a new column, you need to update the column_list and potentially track_history_column_list in the pipeline definition, then do a full refresh. There’s no automatic column discovery for APPLY CHANGES.

Conclusion

#

Hand-coding SCD Type 2 is one of those problems where the basic case is straightforward and the edge cases are brutal. Out-of-order events, duplicate deliveries, late deletes, idempotent replay – each one adds another conditional branch to your merge statement, and the interaction between them creates bugs that only surface in production at the worst possible time.

APPLY CHANGES in Declarative Pipelines compresses all of that into a declaration: here are my keys, here’s my sequence column, here’s what to track. The framework generates the merge logic, handles the edge cases, and produces a clean SCD2 table with __START_AT / __END_AT timestamps that just work.

Pair it with Change Data Feed, and downstream consumers get efficient incremental reads instead of full table scans. That’s a complete CDC-to-SCD2-to-downstream pipeline with maybe 50 lines of actual code. The rest is configuration.