Snowflake表数据变更识别：类型、字段变更及方案咨询

一、现有新增数据查询的优化 #

你当前的新增数据统计逻辑可行，但NOT IN在遇到Cust_Number为NULL时会出现异常结果，推荐改用LEFT JOIN或NOT EXISTS规避这个问题，同时让查询逻辑更清晰：

SELECT 
    'Customer' AS Tablename,
    'Add' AS DeltaType,
    d1.File_Date AS month,
    COUNT(d1.Cust_Number) AS NumRecords
FROM CUSTOMER d1
LEFT JOIN CUSTOMER d2 
    ON d1.Cust_Number = d2.Cust_Number 
    AND d2.File_Date <= '20230724'
WHERE d1.File_Date = '20230822'
    AND d2.Cust_Number IS NULL
GROUP BY d1.File_Date

二、完整变更类型（新增/删除/更新）的实现 #

1. 删除数据统计 #

统计指定时间点被删除的记录，可反向对比历史数据与当前数据：

SELECT 
    'Customer' AS Tablename,
    'Delete' AS DeltaType,
    '20230822' AS month, -- 标记删除发生的时间节点
    COUNT(d2.Cust_Number) AS NumRecords
FROM CUSTOMER d2
LEFT JOIN CUSTOMER d1 
    ON d2.Cust_Number = d1.Cust_Number 
    AND d1.File_Date = '20230822'
WHERE d2.File_Date <= '20230724'
    AND d1.Cust_Number IS NULL
GROUP BY '20230822'

2. 更新数据识别 #

识别更新的记录，可通过主键关联新旧版本数据，对比非主键字段差异。字段较多时，用HASH函数对比整行（排除File_Date）更高效：

SELECT 
    'Customer' AS Tablename,
    'Update' AS DeltaType,
    d1.File_Date AS month,
    COUNT(DISTINCT d1.Cust_Number) AS NumRecords
FROM CUSTOMER d1
JOIN CUSTOMER d2 
    ON d1.Cust_Number = d2.Cust_Number
WHERE d1.File_Date = '20230822'
    AND d2.File_Date = '20230724' -- 对比上一个时间节点的数据
    AND HASH(d1.* EXCLUDE (File_Date)) != HASH(d2.* EXCLUDE (File_Date))
GROUP BY d1.File_Date

三、定位具体变更的字段 #

1. 逐个字段对比（适合字段少的表） #

用CASE WHEN逐个判断字段是否变化，输出变更字段名：

SELECT 
    d1.Cust_Number,
    STRING_AGG(changed_field, ', ') AS changed_fields
FROM (
    SELECT 
        d1.Cust_Number,
        CASE WHEN d1.Name != d2.Name THEN 'Name' END AS changed_field,
        CASE WHEN d1.Phone != d2.Phone THEN 'Phone' END AS changed_field,
        CASE WHEN d1.Email != d2.Email THEN 'Email' END AS changed_field
        -- 继续添加其他需要对比的字段
    FROM CUSTOMER d1
    JOIN CUSTOMER d2 
        ON d1.Cust_Number = d2.Cust_Number
    WHERE d1.File_Date = '20230822'
        AND d2.File_Date = '20230724'
) t
GROUP BY Cust_Number
HAVING changed_fields IS NOT NULL

2. 动态字段对比（适合字段多的表） #

利用Snowflake的OBJECT_CONSTRUCT和LATERAL FLATTEN将行转成键值对，自动对比所有字段：

SELECT 
    d1.Cust_Number,
    STRING_AGG(f.key, ', ') AS changed_fields
FROM CUSTOMER d1
JOIN CUSTOMER d2 
    ON d1.Cust_Number = d2.Cust_Number
,LATERAL FLATTEN(INPUT => OBJECT_CONSTRUCT(* EXCLUDE (File_Date, Cust_Number))) f
,LATERAL FLATTEN(INPUT => OBJECT_CONSTRUCT(d2.* EXCLUDE (File_Date, Cust_Number))) f2
WHERE d1.File_Date = '20230822'
    AND d2.File_Date = '20230724'
    AND f.key = f2.key
    AND f.value != f2.value
GROUP BY d1.Cust_Number

四、推荐最佳方案：Snowflake Streams + Tasks #

如果需要持续自动化捕获数据变更，Snowflake原生的**流（Streams）和任务（Tasks）**是最优选择：

• Streams：自动捕获表的DML变更（INSERT/UPDATE/DELETE），无需手动对比历史数据，记录变更类型和字段值。
• Tasks：定时触发，将Streams捕获的变更同步到专属变更日志表，方便后续分析。

创建流的示例语句：

CREATE OR REPLACE STREAM customer_change_stream
ON TABLE CUSTOMER
APPEND_ONLY = FALSE; -- 设为FALSE可捕获所有DML变更

查询流获取变更详情：

SELECT 
    'Customer' AS Tablename,
    METADATA$ACTION AS DeltaType, -- 输出INSERT/UPDATE/DELETE
    CURRENT_DATE AS month,
    COUNT(*) AS NumRecords,
    -- 定位变更字段，对比BEFORE和AFTER快照
    CASE WHEN METADATA$ACTION = 'UPDATE' THEN 
        ARRAY_TO_STRING(
            ARRAY_COMPARE(OBJECT_CONSTRUCT(BEFORE.*), OBJECT_CONSTRUCT(AFTER.*)), 
            ', '
        ) 
    END AS changed_fields
FROM customer_change_stream
GROUP BY Tablename, DeltaType, month, changed_fields;

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