使用Spark声明式管道与Autoloader进行CDC数据摄入的去重方案咨询

针对每小时CDC数据重复更新的场景，在Databricks声明式管道（Flow）中可通过批次内预去重+声明式UPSERT实现，替代非声明式的foreachBatch+Merge逻辑，具体方案如下：

核心思路 #

1. 批次内去重：读取Autoloader流数据时，对当前批次内的重复记录按ID分区，保留最新时间戳的那条记录。
2. 声明式UPSERT：用APPLY CHANGES INTO语法自动处理目标表的更新/插入操作，无需手动编写Merge逻辑。

具体实现代码 #

1. 创建目标Bronze表（若未存在） #

CREATE TABLE IF NOT EXISTS bronze.test_table (
  id STRING, -- 唯一标识记录的ID列
  col1 STRING, -- 业务字段示例
  col2 INT, -- 业务字段示例
  update_timestamp TIMESTAMP, -- CDC事件的更新时间戳
  -- 其他业务字段
)
USING DELTA
COMMENT "Bronze层存储去重后的CDC增量数据";

2. 创建声明式Flow（含去重与UPSERT逻辑） #

CREATE FLOW bronze.test_table_flow AS
APPLY CHANGES INTO bronze.test_table
FROM STREAM read_files(
  "/Volumes/catalog_dev/bronze/test_table",
  format => "json",
  useManagedFileEvents => 'True',
  singleVariantColumn => 'Data'
)
-- 对当前批次数据去重，保留每个ID的最新记录
WITH deduplicated_data AS (
  SELECT 
    *,
    -- 按ID分区，按更新时间戳降序排序，取第一条
    ROW_NUMBER() OVER (PARTITION BY id ORDER BY update_timestamp DESC) AS rn
  FROM stream
)
SELECT * EXCEPT(rn) -- 移除排序用的辅助字段
FROM deduplicated_data
WHERE rn = 1 -- 仅保留最新记录
-- 定义UPSERT规则：匹配ID时更新全量字段，不匹配时插入
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
;

关键细节说明 #

• 去重逻辑：通过ROW_NUMBER()窗口函数，在每个批次内对相同ID的记录按更新时间戳倒序排列，只保留第一条（最新）记录，避免同一批次内的重复更新。
• APPLY CHANGES INTO：这是Databricks声明式管道中处理CDC场景的原生语法，会自动将流数据与目标表做匹配，执行更新或插入操作，替代了非声明式中手动编写的Merge逻辑。
• Autoloader配置：useManagedFileEvents = 'True'确保Autoloader仅处理新增文件，不会重复读取历史数据，保障增量摄入的正确性。

最佳实践 #

• 确保update_timestamp是CDC事件本身的更新时间，而非文件存储时间，这样才能准确判断记录的新旧。
• 若目标表已有数据，需保证id列是唯一键或主键，避免UPSERT时出现数据不一致。
• 可在目标表启用Delta Lake的变更数据捕获（CDF），方便后续Silver/Gold层基于增量变更做处理：

  ALTER TABLE bronze.test_table SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');
  

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