📚 课程目录第4章缓慢变化维度（SCD）工程

4. 缓慢变化维度（SCD）工程

在维度建模中，客户、产品、门店等实体的属性会随时间变化：客户可能更换手机号，产品可能调整所属品类，门店可能变更店长。缓慢变化维度（Slowly Changing Dimension，SCD）要解决的核心问题不是“如何存下最新值”，而是“当属性变化时，历史事实应当关联到哪个版本”。Snowflake 提供了 MERGE、Stream、Task 三类原语，可以分别承担“合并写入”“捕获变更”“调度执行”的职责，从而把 SCD 从一次性脚本升级为可自动运行的工程链路。

三种 SCD 类型的语义差异

Type 1 采用覆盖策略：新值直接替换旧值，维度表中不保留历史。它适合修正错误数据或对历史追溯无要求的属性，例如客户邮箱拼写纠正。Type 2 采用版本化策略：每次变化插入一条新行，用生效时间、失效时间和当前标志区分版本，历史事实仍指向旧版本代理键。它适合需要“按当时状态分析”的属性，例如客户等级、产品所属品类。Type 3 采用列扩展策略：在同行内增加“当前值”和“先前值”两列，只保留有限历史，适合变化频率低且只需对比上一版的场景，例如门店店长变更。

关键判断：如果业务问题包含“当时是什么”，优先 Type 2；如果只关心“现在是什么”，Type 1 足够；如果只需“和上一版比”，Type 3 成本最低。

Type 1：MERGE 覆盖与幂等重跑

Type 1 的实现最直接。假设已有 DIM_CUSTOMER，源表 STG_CUSTOMER 包含 CUSTOMER_ID、PHONE、EMAIL。使用 MERGE 按业务键匹配，匹配则更新，不匹配则插入：

MERGE INTO DIM_CUSTOMER t
USING STG_CUSTOMER s
ON t.CUSTOMER_ID = s.CUSTOMER_ID
WHEN MATCHED AND (t.PHONE <> s.PHONE OR t.EMAIL <> s.EMAIL) THEN
  UPDATE SET t.PHONE = s.PHONE, t.EMAIL = s.EMAIL, t.UPDATED_AT = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
  INSERT (CUSTOMER_ID, PHONE, EMAIL, UPDATED_AT)
  VALUES (s.CUSTOMER_ID, s.PHONE, s.EMAIL, CURRENT_TIMESTAMP());

这里 WHEN MATCHED AND 的条件很重要：它避免无变化行被反复更新，减少微分区重写。Type 1 的验证方式是检查目标表行数是否等于源表去重后的业务键数量，并抽查被覆盖字段是否与源一致。

Type 2：Stream 捕获变化，MERGE 关闭旧版本并插入新版本

Type 2 的难点在于“先失效旧行，再插入新行”，且必须保证同一业务键在同一时刻只有一个当前版本。Snowflake 的 Stream 可以记录源表上的插入、更新、删除，配合 MERGE 可以做到只处理变化数据。

先建立带版本字段的维度表：

CREATE TABLE DIM_PRODUCT (
  PRODUCT_SK NUMBER AUTOINCREMENT,
  PRODUCT_ID STRING,
  CATEGORY STRING,
  EFFECTIVE_FROM TIMESTAMP_NTZ,
  EFFECTIVE_TO TIMESTAMP_NTZ,
  IS_CURRENT BOOLEAN,
  CHANGE_REASON STRING
);

在源表上创建 Stream：CREATE STREAM STRM_PRODUCT ON TABLE STG_PRODUCT;。随后用一条 MERGE 完成“关旧”和“开新”。由于 MERGE 不能在同一语句中对同一目标表既更新又插入同一业务键的冲突行，工程上通常拆成两步：先用 UPDATE 关闭已变化业务键的当前行，再用 INSERT 写入新版本。Stream 的 METADATA$ACTION 和 METADATA$ISUPDATE 可用于识别变更类型。

-- 步骤一：关闭旧版本
UPDATE DIM_PRODUCT d
SET d.EFFECTIVE_TO = CURRENT_TIMESTAMP(),
    d.IS_CURRENT = FALSE
WHERE d.IS_CURRENT = TRUE
  AND EXISTS (
    SELECT 1 FROM STRM_PRODUCT s
    WHERE s.PRODUCT_ID = d.PRODUCT_ID
      AND s.METADATA$ACTION IN ('INSERT','UPDATE')
      AND s.CATEGORY <> d.CATEGORY
  );

-- 步骤二：插入新版本
INSERT INTO DIM_PRODUCT (PRODUCT_ID, CATEGORY, EFFECTIVE_FROM, EFFECTIVE_TO, IS_CURRENT, CHANGE_REASON)
SELECT PRODUCT_ID, CATEGORY, CURRENT_TIMESTAMP(), NULL, TRUE, 'CATEGORY_CHANGED'
FROM STRM_PRODUCT
WHERE METADATA$ACTION IN ('INSERT','UPDATE');

执行后应检查：每个 PRODUCT_ID 是否恰好有一条 IS_CURRENT = TRUE；历史行的 EFFECTIVE_TO 是否等于新行的 EFFECTIVE_FROM；事实表关联时是否按事件时间落在 [EFFECTIVE_FROM, EFFECTIVE_TO) 区间内。若出现同一业务键多条当前行，说明关闭条件与插入条件不一致，需要回退并按业务键重新对齐。

Type 3：同行双列与有限历史

Type 3 不新增行，而是在维度表中增加 PREV_CATEGORY 列。MERGE 时把当前值写入 PREV_CATEGORY，再更新 CATEGORY：

MERGE INTO DIM_STORE t
USING STG_STORE s
ON t.STORE_ID = s.STORE_ID
WHEN MATCHED AND t.MANAGER <> s.MANAGER THEN
  UPDATE SET t.PREV_MANAGER = t.MANAGER,
             t.MANAGER = s.MANAGER,
             t.UPDATED_AT = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
  INSERT (STORE_ID, MANAGER, PREV_MANAGER, UPDATED_AT)
  VALUES (s.STORE_ID, s.MANAGER, NULL, CURRENT_TIMESTAMP());

Type 3 的局限是只能保留一版历史，第二次变更会覆盖 PREV_MANAGER。因此它更适合变更频率低、且分析只要求“当前 vs 上一版”的属性。

用 Task 自动化调度

Stream 本身不会自动消费，需要 Task 按计划执行。可以创建根任务每 5 分钟运行一次，并在有数据时触发：

CREATE TASK TASK_SCD_PRODUCT
  WAREHOUSE = WH_ETL
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('STRM_PRODUCT')
AS
  CALL SP_MERGE_PRODUCT_SCD2();

把关闭旧版本和插入新版本封装进存储过程，Task 只负责调度，便于失败重试和日志追踪。任务运行后，可通过 SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY()) 查看状态、错误码和耗时。若任务持续报错，优先检查 Stream 是否已被消费、MERGE 条件是否产生多行匹配，以及目标表当前标志是否唯一。

工程化落地要点

第一，代理键生成。Type 2 每次插入新版本都应生成新的 PRODUCT_SK，而业务键 PRODUCT_ID 保持不变，事实表通过代理键关联，才能保证历史事实指向旧版本。第二，时间边界统一采用左闭右开区间 [EFFECTIVE_FROM, EFFECTIVE_TO)，当前行的 EFFECTIVE_TO 置为 NULL 或远未来时间，避免边界重复。第三，Stream 消费具有一次性：Task 成功执行后 Stream 偏移量前移，若需要重放，应使用 CHANGES 或临时表先落地变更集。第四，对 Type 2 的关闭与插入应放在同一事务中，Snowflake 存储过程内可用显式事务保证原子性，防止出现“旧行已关、新行未插”的中间态。

最后，建议为 SCD 链路建立最小监控：统计每次运行的变更行数、当前行唯一性违规数、以及事实表关联不到维度的孤儿行数。当变更行数突增或唯一性违规大于 0 时触发告警，这比事后人工比对维度表更能提前暴露 MERGE 条件写错、Stream 重复消费或时间边界不一致等问题。