Snowflake Streams如何捕获行全部变更而非净变更以构建SCD Type2？

Great question—this is a common pain point when building Type 2 SCDs with Snowflake's default stream behavior, since standard streams optimize for net changes rather than full event history. Your explanation about needing intermediate updates is spot-on: if multiple changes happen to the same row between stream consumes, the standard stream only returns the final state, making it impossible to track all versions required for a complete Type 2 dimension.

Let’s walk through the most practical, native-friendly solutions to capture every change event without writing repetitive Time Travel code:

1. Enable INCLUDE_BEFORE_VALUES for Basic Update/Delete History #

First, when creating your stream, add the INCLUDE_BEFORE_VALUES = TRUE parameter. This makes the stream return both the before and after values for update/delete events—critical for marking old dimension records as expired in Type 2 SCDs.

Example stream creation:

CREATE OR REPLACE STREAM StreamOfMyTable
ON TABLE MyTable
INCLUDE_BEFORE_VALUES = TRUE;

Pros: Simple, native Snowflake feature, no extra infrastructure needed.
Cons: Still returns net changes (one record per row with the latest state). If you need to capture every intermediate update (e.g., three edits to the same row between stream runs), this alone won’t solve it—but it’s a foundational step for any SCD work.

2. Build a Centralized Change Log Table + Automated Capture Procedure #

To capture every single change event (including intermediate updates), create a dedicated audit table and use a stored procedure to automate pulling both stream data and missed intermediate changes via Time Travel. This removes the need to write custom Time Travel queries for every table.

Step 1: Create the Change Log Table

This table will store every modification event, including before/after values and timestamps:

CREATE OR REPLACE TABLE MyTable_Change_Log (
    change_id INT AUTOINCREMENT PRIMARY KEY,
    change_timestamp TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
    change_type VARCHAR(10) NOT NULL, -- INSERT/UPDATE/DELETE
    before_data VARIANT, -- Old values for updates/deletes
    after_data VARIANT, -- New values for inserts/updates
    source_row_id INT NOT NULL -- Maps back to MyTable's primary key
);

Step 2: Write a Stored Procedure to Capture Full Changes

This procedure will:

• Track the last time you consumed changes from the source table
• Use Time Travel to pull all changes since that last run
• Log everything to the change log table
• Advance the stream offset to avoid reprocessing data

CREATE OR REPLACE PROCEDURE Capture_Full_MyTable_Changes()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    last_consume_time TIMESTAMP_NTZ;
    current_run_time TIMESTAMP_NTZ := CURRENT_TIMESTAMP();
BEGIN
    -- Get the most recent change capture time from our log
    SELECT MAX(change_timestamp) INTO last_consume_time FROM MyTable_Change_Log;
    
    -- If no prior runs, start from the stream's creation date
    IF last_consume_time IS NULL THEN
        last_consume_time := (SELECT created FROM INFORMATION_SCHEMA.STREAMS WHERE stream_name = 'STREAMOFMYTABLE');
    END IF;
    
    -- Capture all changes (inserts/updates/deletes) since the last run
    INSERT INTO MyTable_Change_Log (change_type, before_data, after_data, source_row_id)
    SELECT
        metadata$action AS change_type,
        metadata$before AS before_data,
        metadata$after AS after_data,
        id AS source_row_id
    FROM MyTable AT(OFFSET => -1) -- Uses Time Travel; adjust offset based on your retention policy
    WHERE metadata$action IN ('INSERT', 'UPDATE', 'DELETE')
      AND metadata$commit_time BETWEEN last_consume_time AND current_run_time;
    
    -- Mark stream changes as consumed by selecting all records into a temp table
    CREATE OR REPLACE TEMPORARY TABLE temp_stream_consume AS SELECT * FROM StreamOfMyTable;
    
    RETURN 'Successfully captured ' || SQLROWCOUNT || ' change events.';
END;
$$;

Step 3: Schedule with a Task

Run this procedure on a frequent schedule (e.g., every 5 minutes) to ensure you capture intermediate updates before they get merged into the stream’s net change:

CREATE OR REPLACE TASK Capture_MyTable_Changes_Task
WAREHOUSE = Your_Warehouse_Name
SCHEDULE = '5 MINUTE'
AS
CALL Capture_Full_MyTable_Changes();

ALTER TASK Capture_MyTable_Changes_Task RESUME;

Pros: Captures every single change event, including intermediate updates. Automates the Time Travel logic so you don’t have to write it per table.
Cons: Requires maintaining the audit table and stored procedure, but this is a one-time setup per table.

3. Use Append-Only Streams (Insert-Heavy Workloads) #

If your workload is mostly inserts (with rare updates/deletes), an append-only stream will capture every insert event without merging. This is a lightweight option but won’t track updates or deletes—so pair it with the change log approach above if you need full history.

Example append-only stream:

CREATE OR REPLACE STREAM StreamOfMyTable_Append
ON TABLE MyTable
APPEND_ONLY = TRUE;

4. Leverage Source-Side CDC (If Available) #

If your source system (e.g., MySQL, PostgreSQL, SQL Server) supports native CDC, ingest the full change history directly into Snowflake using Snowpipe or External Tables. This bypasses Snowflake’s stream net change limitation entirely, as you’re capturing every event at the source before it reaches Snowflake.

Final Tip for SCD Type 2 #

Once you have your full change log, building the Type 2 dimension becomes straightforward:

• For insert events: Add a new dimension record with start_date set to the change timestamp and end_date as NULL.
• For update events: Mark the existing dimension record (matching the source row ID) with end_date set to the change timestamp, then add a new record with the updated values and start_date set to the change timestamp.

内容的提问来源于stack exchange，提问作者Gavin Wilson