Snowflake中CDC流多更新场景下SCD2维度表合并方案问询

核心思路 #

当源数据中同一ID的多份变更记录同时到达时，需先对增量数据进行有序化、去重预处理，再通过分步骤操作（先过期旧版本，再插入所有有效新版本）实现完整的SCD Type 2历史记录保留，避免单条MERGE无法覆盖多版本的问题。

方案一：处理单ID多版本同时到达的完整历史记录 #

该方案适用于需要保留同一ID所有变更版本的场景，即使这些版本同时批量到达。

步骤1：预处理增量数据 #

对STG层数据按ID和变更时间排序，生成每个版本的生效/失效时间，并过滤重复记录：

WITH ordered_staged AS (
    SELECT 
        id,
        name,
        email,
        load_timestamp,
        -- 获取当前记录的下一个变更时间，作为自身的失效时间
        LEAD(load_timestamp) OVER (PARTITION BY id ORDER BY load_timestamp, METADATA$FILE_ROW_NUMBER) AS next_change_time
    FROM STG_PEOPLE
    -- 过滤完全重复的记录（同ID、同时间、同属性）
    QUALIFY ROW_NUMBER() OVER (PARTITION BY id, load_timestamp, name, email) = 1
),
staged_versioned AS (
    SELECT 
        id,
        name,
        email,
        load_timestamp AS start_date,
        -- 最后一个版本的失效时间设为远未来
        COALESCE(next_change_time, '9999-12-31'::TIMESTAMP) AS end_date,
        -- 标记当前有效版本
        IFF(next_change_time IS NULL, TRUE, FALSE) AS is_current
    FROM ordered_staged
)
SELECT * FROM staged_versioned;

步骤2：过期维度表中旧的当前版本 #

先将DIM_PEOPLE中对应ID的当前有效版本标记为过期：

UPDATE DIM_PEOPLE dp
SET 
    end_date = (SELECT MIN(start_date) FROM staged_versioned sv WHERE sv.id = dp.id) - INTERVAL '1 second',
    is_current = FALSE
WHERE 
    dp.is_current = TRUE
    AND EXISTS (SELECT 1 FROM staged_versioned sv WHERE sv.id = dp.id);

步骤3：插入所有新的版本记录 #

将预处理后的多版本记录插入维度表，避免重复插入已有版本：

INSERT INTO DIM_PEOPLE (id, name, email, start_date, end_date, is_current)
SELECT 
    id, 
    name, 
    email, 
    start_date, 
    end_date, 
    is_current
FROM staged_versioned
WHERE NOT EXISTS (
    SELECT 1 FROM DIM_PEOPLE dp
    WHERE 
        dp.id = staged_versioned.id
        AND dp.start_date = staged_versioned.start_date
        AND dp.name = staged_versioned.name
        AND dp.email = staged_versioned.email
);

方案二：仅处理单ID最新变更（增量更新） #

如果业务只需要保留最新变更，无需回溯历史版本，可使用带排序的MERGE语句简化操作：

MERGE INTO DIM_PEOPLE dp
USING (
    SELECT 
        id,
        name,
        email,
        load_timestamp,
        -- 取每个ID的最新有效变更（按时间+文件行号排序）
        ROW_NUMBER() OVER (PARTITION BY id ORDER BY load_timestamp DESC, METADATA$FILE_ROW_NUMBER DESC) AS rn
    FROM STG_PEOPLE
    -- 过滤掉与当前维度表最新记录完全一致的无效变更
    WHERE NOT EXISTS (
        SELECT 1 FROM DIM_PEOPLE dp
        WHERE 
            dp.id = STG_PEOPLE.id
            AND dp.is_current = TRUE
            AND dp.name = STG_PEOPLE.name
            AND dp.email = STG_PEOPLE.email
    )
) stg
ON dp.id = stg.id AND dp.is_current = TRUE
WHEN MATCHED THEN
    UPDATE SET 
        dp.end_date = CURRENT_TIMESTAMP(),
        dp.is_current = FALSE
WHEN NOT MATCHED THEN
    INSERT (id, name, email, start_date, end_date, is_current)
    VALUES (stg.id, stg.name, stg.email, stg.load_timestamp, '9999-12-31'::TIMESTAMP, TRUE);

关键注意事项 #

• 时间字段准确性：确保load_timestamp反映数据的实际变更时间，而非Snowpipe的摄入时间；若依赖摄入时间，使用METADATA$FILE_TIMESTAMP替代。
• 排序逻辑：使用METADATA$FILE_ROW_NUMBER或METADATA$FILENAME辅助排序，确保同时间戳的记录按摄入顺序处理。
• 重复过滤：通过QUALIFY或NOT EXISTS过滤无效重复记录，避免维度表产生冗余版本。

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