Use SCDs for effective history management

In the Data world, pipelines and tables can be defined by some pairs of concepts.

One important distinction is the rate of change. Some tables change slowly, and some tables change rapidly.

Also, another pair of concepts we can classify tables with is: dimensional or non-dimensional.

Each type of table brings along pipeline-related concerns.

Let's say you're extracting and processing application logs. You can define your grain as each individual line of the log. This type of processing is mostly non-dimensional and append-only, and each grain never changes. So we have never-changing facts.

On the other hand, we can have a dimensional modelling with fast change, like a product availability pipeline.

In this pipeline, your grain will be the latest status of a given product. Since this changes a lot, this pipeline shouldn't return many duplicates. You can have two tables: latest_snapshot and history.

History can be append-only, and the latest snapshot is a full overwrite.

But what if you have a table that's dimensional and changes slowly? Say a master table with employee data?

If you apply your strategy for the product availability pipeline, your history table will be polluted with repeated data. After all, your employees won't change salaries, addresses, or job titles daily.

What if there's a way to have a table with historical and latest values together?

You can do this with modelling for slowly changing dimensions.

In the industry, there are three ways of maintaining an SCD workflow. These Oracle documents provide a TL;DR version of these formulas.

Let's deep dive into the types of SCDs using an employee table as example. Here's the base schema.

Note that we'll be using MERGE INTO statements, so you'll want a processing engine that is compatible with these features. Our code will be ANSI-compliant.

CREATE TABLE employee_master as (
   ID                VARCHAR
   NAME              VARCHAR,
   JOB_TITLE         VARCHAR,
   MANAGER_NAME      VARCHAR,
   MANAGER_ID        VARCHAR,
   DEPARTMENT_ID     VARCHAR,
   DEPARTMENT_NAME   VARCHAR,
   SNAPSHOT_DATE     DATE
)

This is the fastest and most brute-force way of handling these records. As seen in the Oracle docs, only one version of the dimension record exists. When a change is made, the record is overwritten.

I say this is a brute-force approach because no history is preserved.

So, if your history is being preserved elsewhere or you can afford to disregard it entirely, this is your use case.

Here's a base SQL statement for this.

-- when matched by ID (regardless of snapshot_date), insert everything
-- otherwise, insert, because it's a new record
MERGE INTO employee_master AS target
USING incoming_employee_data AS source
ON target.id = source.id
WHEN MATCHED THEN
  UPDATE SET
    name = source.name,
    job_title = source.job_title,
    manager_name = source.manager_name,
    manager_id = source.manager_id,
    department_id = source.department_id,
    department_name = source.department_name,
    snapshot_date = CURRENT_DATE
WHEN NOT MATCHED THEN
  INSERT (id, name, job_title, manager_name, manager_id, department_id, department_name, snapshot_date)
  VALUES (source.id, source.name, source.job_title, source.manager_name, source.manager_id, source.department_id, source.department_name, CURRENT_DATE);

In this type of SCD, multiple versions of the same entity can be present. To enable this workflow, you'll need a column containing a validity flag.

This column can be a boolean like is_valid, or you can have two date columns like created_at and expired_at.

To write on this table, you'll need a transaction with two operations: one to update the expired records and another to insert the most recent rows.

This is a good alternative if your downstream users are not SQL heavy, because it's very straightforward to simply filter the table to return historical or current records.

Let's get to the code:

Using a validity flag

-- step 1: expire the old version
-- (it assumes that is_valid comes from source)
UPDATE employee_master
SET is_valid = FALSE
WHERE id IN (SELECT id FROM incoming_employee_data)
  AND is_valid = TRUE;
-- step 2: insert the new version
INSERT INTO employee_master (
  id, name, job_title, manager_name, manager_id,
  department_id, department_name, snapshot_date, is_valid
)
SELECT
  id, name, job_title, manager_name, manager_id,
  department_id, department_name, CURRENT_DATE, TRUE
FROM incoming_employee_data;

Using expire date

-- step 1: expire the previous version
UPDATE employee_master
SET expired_at = CURRENT_DATE
WHERE id IN (SELECT id FROM incoming_employee_data)
  AND expired_at IS NULL;
-- step 2: insert new version
INSERT INTO employee_master (
  id, name, job_title, manager_name, manager_id,
  department_id, department_name, created_at, expired_at
)
SELECT
  id, name, job_title, manager_name, manager_id,
  department_id, department_name, CURRENT_DATE, NULL
FROM incoming_employee_data;

In this last type of SCD, you'll have one row per entity. In this row, you'll have one with the record's past values among the normal columns. This column must contain an object, like an array of structs, containing the past values.

This type of SCD is great for efficient storage. You'll potentially compress tens of historical records within a single column.

However, this is not a good alternative if your downstream processes are not very keen on SQL, since it's not straightforward to query your nested historical columns.

Here's the code:

MERGE INTO employee_master AS target
USING incoming_employee_data AS source
ON target.id = source.id
-- existing record
WHEN MATCHED THEN
  UPDATE SET
    job_title = source.job_title,
    snapshot_date = CURRENT_DATE, -- or date from source
    history = ARRAY_CONCAT(
      target.history,
      ARRAY[STRUCT(target.job_title, target.snapshot_date)] -- cols you want to store
    )
-- new record
WHEN NOT MATCHED THEN
  INSERT (
    id, name, job_title, manager_name, manager_id,
    department_id, department_name, snapshot_date, history
  )
  VALUES (
    source.id, source.name, source.job_title, source.manager_name,
    source.manager_id, source.department_id, source.department_name,
    CURRENT_DATE,
    ARRAY[] -- no history yet for new record
  );

I personally like these types of workflows a lot.

Just note that, to use them with PySpark, you'll need a lakehouse environment. Then, your tables will be MERGE-compatible, with engines like Delta or Iceberg.

I hope you liked it and learned something new!