Slowly Changing Dimensions in ADF Data Flows

A customer moves from Rio de Janeiro to Pune. Should last year's sales still be reported under Rio? If yes, the dimension has to remember both addresses and when each applied. That is a slowly changing dimension, and loading one correctly is the classic test of a transformation tool. This guide builds the load as a mapping data flow in Azure Data Factory.

For the types themselves and when to use each, see slowly changing dimensions explained. In brief: Type 1 overwrites and keeps no history; Type 2 adds a new row for each change.

What a Type 2 dimension needs

A Type 2 change adds a row, so one customer can have several rows. The Kimball Group's description gives the table its shape (Kimball Group: Type 2):

• A surrogate key, because the business key no longer identifies one row.
• A row effective date and a row expiration date.
• A current row indicator.

One more column makes the load simpler: a hash of the tracked attributes, so "did anything change?" is a single comparison.

Type 1 first: one transformation

Type 1 is an upsert. The flow is a source, an Alter row with one policy, and a database sink:

• Alter row marks every row for upsert, with the condition true().
• The sink allows upsert and names the business key as its key column.

That is all. The rest of this article is about Type 2.

The Type 2 flow, stream by stream

Unchanged rows go nowhere. In data flow script, steps 2, 4, 5 and 6 look like this:

StagedCustomers derive(RowHash = md5(City, Country, Company)) ~> AddHash
AddHash, CurrentDim lookup(AddHash@CustomerId == CurrentDim@CustomerId,
    multiple: false,
    pickup: 'any',
    broadcast: 'auto') ~> FindCurrent
FindCurrent split(isNull(CustomerKey),
    RowHash != DimRowHash,
    disjoint: false) ~> Route@(NewRows, ChangedRows, Unchanged)
ChangedRows alterRow(updateIf(true())) ~> MarkExpire

Notes on each step:

The hash. md5 takes a set of columns and returns a 32-character string, which Microsoft describes as usable as a fingerprint for a row (Microsoft Learn: data flow functions). Hash only the columns whose history you want. A change to an untracked column should not create a version.

The lookup. It keeps every incoming row and adds the dimension's columns where the key matches. Match multiple rows must be off: one current row per customer is the rule, and a lookup that returns two would double the insert (Microsoft Learn: Lookup). The dimension source should be filtered to current rows in its query, so history is never read.

The split. Rows go to the first condition they match: no key found means new; a different hash means changed; everything else falls to the default stream (Microsoft Learn: Conditional split).

The two sinks. The expiring sink allows update, with the surrogate key as its key column, and maps the end date and the current flag. The insert sink only inserts. A database sink inserts by default and acts on updates only when an Alter row has marked the rows and the sink's settings allow it (Microsoft Learn: Alter row).

The same logic, run four times

The script below reproduces the flow in SQLite on the Customer table of the Chinook sample database (Chinook on GitHub), tracking city, country and company. Between loads it changes the source: two customers move, one company changes, one phone number changes, one customer is added, and later one customer moves back. It is a local reproduction of the logic, not Data Factory.

import hashlib, shutil, sqlite3

shutil.copy("chinook.sqlite", "work.sqlite")                 # never alter the downloaded file
db = sqlite3.connect("work.sqlite")
# the role md5(...) plays in a derived column: one fingerprint for the tracked columns
db.create_function("row_hash", -1, lambda *cols: hashlib.md5(
    "|".join("" if c is None else str(c) for c in cols).encode()).hexdigest())
db.executescript("""
    CREATE TABLE dim_customer (
        CustomerKey INTEGER PRIMARY KEY,     -- surrogate key
        CustomerId  INTEGER,                 -- business key
        City TEXT, Country TEXT, Company TEXT,
        RowHash TEXT, StartDate TEXT, EndDate TEXT, IsCurrent INTEGER);
""")


def load(run_date):
    # Source + Derived column (hash), Lookup against the current dimension rows, Conditional split
    db.executescript("DROP TABLE IF EXISTS temp.routed;")
    db.execute("""
        CREATE TEMP TABLE routed AS
        SELECT s.CustomerId, s.City, s.Country, s.Company,
               row_hash(s.City, s.Country, s.Company) AS RowHash, d.CustomerKey AS OldKey,
               CASE WHEN d.CustomerKey IS NULL THEN 'new'
                    WHEN d.RowHash <> row_hash(s.City, s.Country, s.Company) THEN 'changed'
                    ELSE 'unchanged' END AS Stream
        FROM Customer s
        LEFT JOIN dim_customer d ON d.CustomerId = s.CustomerId AND d.IsCurrent = 1""")
    n = dict(db.execute("SELECT Stream, COUNT(*) FROM routed GROUP BY Stream"))
    # Sink 1, Alter row = update: close the current version of each changed row
    db.execute("""UPDATE dim_customer SET EndDate = ?, IsCurrent = 0
                  WHERE CustomerKey IN (SELECT OldKey FROM routed WHERE Stream = 'changed')""", (run_date,))
    # Sink 2, insert: new rows and new versions; Surrogate key continues from the current maximum
    max_key = db.execute("SELECT COALESCE(MAX(CustomerKey), 0) FROM dim_customer").fetchone()[0]
    db.execute("""
        INSERT INTO dim_customer
        SELECT ? + ROW_NUMBER() OVER (ORDER BY CustomerId), CustomerId, City, Country, Company, RowHash, ?, NULL, 1
        FROM routed WHERE Stream IN ('new', 'changed')""", (max_key, run_date))
    db.commit()
    rows, current = db.execute("SELECT COUNT(*), SUM(IsCurrent) FROM dim_customer").fetchone()
    print(f"{run_date}  new {n.get('new', 0):2}  changed {n.get('changed', 0):2}  unchanged {n.get('unchanged', 0):2}"
          f"   dimension rows {rows:2}, current {current:2}")


load("2024-01-01")                                           # first load

db.executescript("""
    UPDATE Customer SET City = 'Pune', Country = 'India' WHERE CustomerId = 12;   -- moved
    UPDATE Customer SET City = 'Lyon' WHERE CustomerId = 40;                      -- moved
    UPDATE Customer SET Company = 'Contoso Ltd' WHERE CustomerId = 20;            -- tracked column
    UPDATE Customer SET Phone = '+1 (514) 000-0000' WHERE CustomerId = 3;         -- not a tracked column
    INSERT INTO Customer (CustomerId, FirstName, LastName, City, Country, Email)
        VALUES (60, 'Asha', 'Rao', 'Hyderabad', 'India', 'asha.rao@example.com');
""")
load("2024-02-01")
load("2024-03-01")                                           # nothing changed at the source

db.execute("UPDATE Customer SET City = 'Rio de Janeiro', Country = 'Brazil' WHERE CustomerId = 12")   # moved back
load("2024-04-01")

print("\nhistory of customer 12:")
for row in db.execute("""SELECT CustomerKey, City, Country, StartDate, COALESCE(EndDate, '-'), IsCurrent
                         FROM dim_customer WHERE CustomerId = 12 ORDER BY CustomerKey"""):
    print("  key {:2}  {:15} {:7} from {}  to {:10}  current {}".format(*row))

city = db.execute("""SELECT City FROM dim_customer WHERE CustomerId = 12
                     AND StartDate <= '2024-02-15' AND (EndDate > '2024-02-15' OR EndDate IS NULL)""").fetchone()[0]
print("city of customer 12 on 2024-02-15:", city)

2024-01-01  new 59  changed  0  unchanged  0   dimension rows 59, current 59
2024-02-01  new  1  changed  3  unchanged 56   dimension rows 63, current 60
2024-03-01  new  0  changed  0  unchanged 60   dimension rows 63, current 60
2024-04-01  new  0  changed  1  unchanged 59   dimension rows 64, current 60

history of customer 12:
  key 12  Rio de Janeiro  Brazil  from 2024-01-01  to 2024-02-01  current 0
  key 60  Pune            India   from 2024-02-01  to 2024-04-01  current 0
  key 64  Rio de Janeiro  Brazil  from 2024-04-01  to -           current 1
city of customer 12 on 2024-02-15: Pune

Each line is worth reading:

• First load: 59 new. An empty dimension makes every row new.
• Second load: 1 new, 3 changed, 56 unchanged. Four source rows were updated, and only three are changes. The phone number is not a tracked column, so customer 3 gets no new version. The dimension grows by four rows to 63, and 60 are current, one per customer.
• Third load: nothing. No source changes, no new rows. A Type 2 load that adds rows when nothing has changed is broken, and this is the check that proves it is not. Idempotency in data pipelines explains why that property matters.
• Fourth load: the customer moves back. The hash now equals the first version's hash, and a third version is still created, correctly. The comparison is with the current row only.

The history of customer 12 shows three keys with dates that meet end to start, and the last query answers the question the dimension exists for: on 15 February the customer was in Pune.

Surrogate keys

The Surrogate key transformation numbers rows from a start value. On the second load it must continue from the highest key in the table. Microsoft's documented approach: read the maximum with a source query, write it to a cache sink, and add it to the generated number in a derived column (Microsoft Learn: Surrogate key). The reproduction does the same with MAX(CustomerKey).

The alternative is to let the database assign the key from an identity column and leave that column unmapped in the insert sink. It is simpler, and it moves key generation out of the flow.

Details that decide whether it works

Run it from a pipeline. Alter row policies are not applied in a debug session. Data preview shows which rows are marked; nothing is updated until a pipeline runs the flow.

Sink order. With two sinks on one table, the order of writing is not fixed unless you enable custom sink ordering (Microsoft Learn: Sink). Order the expiring sink first, so the table never holds two current rows for a customer.

Pass the run date in. Use a data flow parameter set by the pipeline for the effective date, not the cluster's clock. A rerun for the same date then writes the same dates.

Duplicates in the source. Two incoming rows for one customer both look "changed" and both insert. Remove duplicates before the lookup, with an Aggregate on the business key.

Deletes. A customer missing from the source matches nothing in this flow, so its row stays current for ever. To close such rows, add an Exists transformation set to "doesn't exist", from the dimension to the source, and route its output to the expiring sink. Do this only when the source is a full snapshot; with an incremental extract, almost every customer is "missing".

Several changes between loads. Comparing snapshots captures the net change. If a customer moved twice since the last run, only the latest city is recorded. Capturing every change needs change data capture at the source.

Backdated changes. This flow dates each version from the load date. If the business needs the date the change really happened, the logic for inserting a version into the middle of a history is harder; see handling late-arriving data.

Mixing types

Real dimensions mix types: history for city, overwrite for a corrected spelling of a name. Hash the two groups separately. A change in the Type 2 hash follows the flow above; a change only in the Type 1 hash goes to an update of the current row. Designing fact and dimension tables covers choosing the type attribute by attribute, and the transformation reference describes each transformation used here.

Azure Data Factory and dimensional modelling for data warehouses are covered in the Data Engineering course.

Sources

• Type 2: Add New Row (Kimball Group) — Kimball Group
• Alter row transformation in mapping data flow (Microsoft Learn) — Microsoft Learn
• Lookup transformations in mapping data flow (Microsoft Learn) — Microsoft Learn
• Conditional split transformation in mapping data flow (Microsoft Learn) — Microsoft Learn
• Surrogate key transformation in mapping data flow (Microsoft Learn) — Microsoft Learn
• Sink transformation in mapping data flow (Microsoft Learn) — Microsoft Learn
• Details and usage for all mapping data flow functions (Microsoft Learn) — Microsoft Learn
• Chinook sample database (GitHub) — GitHub