缓慢变化维度（SCD）

概念定义

业务属性会漂移：客户换分层、商品换类目。关于上季度的报表必须看到上季度的取值。SCD type 2 把每次值变更做成新的维度行，配独立代理键与生效窗口，三月发生的事实就连接三月的版本。

工作机制

管道检测属性变化、关闭当前行、插入带新生效时间戳的继任行；当前行标志或最大日期标记方便最新值查询。晚到事实按有效期窗口连接而非加载时间；晚到的维度变更需要追溯性重版本化。

使用场景

任何监管或分析上要求复现历史分组的场景：按“当时上报的分层”看收入、按“注册时区域”看流失。属性历史无业务含义时用 type 1 覆盖即可。

常见误区

type 2 不是数据类型而是行版本化策略，须刻意设计键与日期。事实必须按“事实发生时当前版本”的代理键连接，按自然键连接会把全部历史塌缩成最新属性。高频变更属性整行复制会造成行数膨胀，应考虑微型维度；回填时代理键生成也要保持稳定。

代码示例

import csv
import io


def detect_dialect(sample: str) -> dict[str, object]:
    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|")
    return {
        "delimiter": dialect.delimiter,
        "quotechar": dialect.quotechar,
        "doublequote": dialect.doublequote,
        "skipinitialspace": dialect.skipinitialspace,
    }


def read_rows(text: str) -> list[dict[str, str]]:
    dialect = csv.Sniffer().sniff(text, delimiters=",;\t|")
    reader = csv.DictReader(io.StringIO(text), dialect=dialect)
    return [dict(row) for row in reader]


if __name__ == "__main__":
    sample = "name;role;city\nAda;engineer;London\nGrace;admiral;New York\n"
    print(detect_dialect(sample))
    print(read_rows(sample))

-- 每个邮箱保留 created_at 最新的一行，标记其余为重复
WITH ranked AS (
  SELECT
    id,
    email,
    created_at,
    ROW_NUMBER() OVER (
      PARTITION BY email
      ORDER BY created_at DESC, id DESC
    ) AS rn
  FROM subscribers
)
SELECT id, email, created_at, (rn = 1) AS is_primary
FROM ranked
ORDER BY email, rn;

-- 一次性清扫：删除全部非首行
WITH ranked AS (
  SELECT id,
         ROW_NUMBER() OVER (
           PARTITION BY email
           ORDER BY created_at DESC, id DESC
         ) AS rn
  FROM subscribers
)
DELETE FROM subscribers s
USING ranked r
WHERE s.id = r.id
  AND r.rn > 1;

-- 上游堵漏：唯一索引阻止重复再次写入
CREATE UNIQUE INDEX CONCURRENTLY idx_subscribers_email
  ON subscribers (email);