SCD Type 2 History with stored_as_scd_type in AUTO CDC

1.Switching a flow to SCD Type 2

An AUTO CDC flow writes either an in-place SCD Type 1 table or an SCD Type 2 history table, and one parameter decides which: stored_as_scd_type in Python, STORED AS in SQL. Set it to 1 for Type 1 or 2 for Type 2; the parameter accepts a string or an int. Type 2 retains a history of records. That can be on all updates, or only on updates to a set of columns you choose.

Instead of overwriting, Type 2 adds a new row for each version of a record. Two columns record when each version was active: __START_AT and __END_AT. Their values come from your sequencing column. In the flow below, the source view, keys, sequencing and delete condition are the standard documentation example. What makes it Type 2 is stored_as_scd_type = "2" and the absence of a truncate condition.

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE  ? ;

Sources12

2.Reading the history: __START_AT, __END_AT and deletes

The documentation's sample change feed main.cdc_tutorial.users_cdf has four users with an operation column and a sequenceNum. Isabel (123) is inserted in Monterrey at sequence 1, updated to Chihuahua at 5, and deleted at 6. Mercedes (125) is inserted in Tijuana at 2, updated to Mexicali at 5, and updated to Guadalajara at 6. The Mexicali update arrives after the Guadalajara one. Raul (124) and Lily (126) are inserted and never change.

The late Mexicali update isn't thrown away. It appears as a real version (5 to 6), placed in sequence order even though it arrived last. That is the difference from an in-place Type 1 table, which keeps only Mercedes' latest city. A row with __END_AT = null is the current version. A deleted key has no open row, only closed history. To get today's view of the dimension, filter on __END_AT IS NULL.

The column names are fixed. If downstream consumers expect names like valid_from and valid_to, the documented approach is to create a view over the target that aliases __START_AT and __END_AT.

Sources23

3.Type 2 rules: selective history, truncates and schemas

By default every change to any output column opens a new version. That is equivalent to TRACK HISTORY ON *. When some columns change often but don't matter historically, limit which columns generate history with track_history_column_list / track_history_except_column_list in Python, or TRACK HISTORY ON in SQL. In the example below, a change to city alone does not create a history record.

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW flow
AS AUTO CDC INTO
  target
FROM stream(cdc_data.users)
  KEYS (userId)
  APPLY AS DELETE WHEN operation = "DELETE"
  SEQUENCE BY sequenceNum
  COLUMNS * EXCEPT (operation, sequenceNum)
  STORED AS SCD TYPE 2
  TRACK HISTORY ON * EXCEPT (city);

Two rules from the reference are easy to miss.

- No truncates in Type 2. apply_as_truncates / APPLY AS TRUNCATE WHEN work only with Type 1. That's why the Type 2 example leaves the truncate condition out. - Explicit schemas must include the history columns. You can optionally declare the target table's schema yourself. If you do for a Type 2 target, it must include __START_AT and __END_AT with the same data type as the sequence_by field. So if you sequence by a timestamp, those columns are timestamps too.

There is a third value for stored_as_scd_type: "bitemporal" (STORED AS BITEMPORAL), which is in Beta. It extends Type 2 with a second timeline for system time. It needs system_sequence_by / SYSTEM SEQUENCE BY and adds __SYSTEM_START_AT and __SYSTEM_END_AT columns. The exam objective is about Types 1 and 2; just recognise this value so it doesn't catch you out.

Sources41

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.A DELETE in an SCD Type 2 flow removes all of the key's rows from the target.Why is that wrong?

   The delete closes the active version by setting its __END_AT. Earlier versions stay, and the key simply has no row with __END_AT null.

   Covered in Reading the history: __START_AT, __END_AT and deletes

2. 2.APPLY AS TRUNCATE WHEN can be added to any AUTO CDC flow, including SCD Type 2.Why is that wrong?

   Truncate handling is supported only for SCD Type 1; SCD Type 2 does not support it.

   Covered in Type 2 rules: selective history, truncates and schemas

3. 3.When you declare an explicit schema for an SCD Type 2 target, you list only your business columns and the pipeline adds __START_AT and __END_AT.Why is that wrong?

   An explicit Type 2 schema must include __START_AT and __END_AT, with the same data type as the sequencing column.

   Covered in Type 2 rules: selective history, truncates and schemas

Practise it for real

Run the documentation's sample change feed through an SCD Type 1 and an SCD Type 2 AUTO CDC flow and compare the two targets.

1. 1.From your pipeline's exploration folder (not the transformations folder), run the sample-data SQL that creates main.cdc_tutorial.users_cdf with userId, name, city, operation and sequenceNum.

   Why: The sample data is not meant to be part of the pipeline definition. It is the CDC feed the flows read from.

   You should see: A table with eight change rows: inserts at sequence 1–2, a DELETE for 123 at 6, and updates at 5 and 6 with the sequence-5 rows listed last.

2. 2.In the transformations folder, add the SCD Type 1 Python code: the users view, create_streaming_table("users_current"), and create_auto_cdc_flow with stored_as_scd_type = 1. Run the pipeline.

   Why: This shows the default upsert behaviour and how out-of-order events are resolved by sequenceNum.

   You should see: users_current holds three rows: 124 Oaxaca, 125 Guadalajara, 126 Cancun.

3. 3.Add a second target, users_history, with create_auto_cdc_flow and stored_as_scd_type = "2", leaving out apply_as_truncates. Run the pipeline again.

   Why: Type 2 keeps every version, and truncate handling is not supported for it.

   You should see: users_history holds seven rows with __START_AT and __END_AT. 124, 125/Guadalajara and 126 have __END_AT null, and user 123 has two closed versions.

4. 4.Query users_history filtered on __END_AT IS NULL and compare the result with users_current.

   Why: The current slice of a Type 2 table should match the Type 1 table for the same feed.

   You should see: The same three users and cities as users_current.