构建Delta Live Table实现Bronze到Silver层增量数据合并

核心逻辑 #

Delta Live Table（DLT）通过内置的增量跟踪和变更合并能力，能高效实现Bronze层原始数据到Silver层转换后数据的持续同步。核心步骤包括：

• 识别Bronze层表x的增量写入数据
• 对原始数据进行清洗转换
• 基于唯一键将增量数据合并到Silver层表Y中

SQL 实现方案 #

场景1：仅新增数据（无更新/删除） #

如果Bronze层仅写入新增数据，无需处理更新，可直接基于增量读取构建Silver层表：

CREATE LIVE TABLE silver_table_y
COMMENT "Silver层转换后数据存储表"
TBLPROPERTIES ("quality" = "silver")
AS
SELECT 
  id, -- 业务唯一键
  CAST(raw_event_time AS TIMESTAMP) AS event_time, -- 类型转换
  TRIM(raw_user_input) AS cleaned_input, -- 字段清洗
  CURRENT_TIMESTAMP() AS silver_ingestion_time
FROM live.bronze_table_x
-- 仅读取上次同步后新增的数据
WHERE _commit_timestamp > (SELECT COALESCE(MAX(silver_ingestion_time), '1970-01-01') FROM live.silver_table_y);

场景2：需处理更新/删除（CDC场景） #

如果Bronze层包含数据的更新、删除操作，使用APPLY CHANGES INTO语法实现全量变更合并：

-- 先定义Silver层目标表结构
CREATE LIVE TABLE silver_table_y
COMMENT "Silver层转换后数据存储表"
TBLPROPERTIES ("quality" = "silver")
USING DELTA
AS SELECT 
  id,
  TIMESTAMP '1970-01-01' AS event_time,
  '' AS cleaned_input,
  CURRENT_TIMESTAMP() AS silver_ingestion_time
LIMIT 0;

-- 配置增量合并规则
APPLY CHANGES INTO live.silver_table_y
FROM live.bronze_table_x
KEYS (id) -- 指定业务唯一键用于匹配记录
SEQUENCE BY _commit_timestamp -- 按Bronze层数据提交时间确保处理顺序
COLUMNS * EXCEPT (_commit_version, _commit_timestamp) -- 排除DLT内部元数据字段
WHEN MATCHED THEN UPDATE SET * -- 匹配到则更新全量字段
WHEN NOT MATCHED THEN INSERT *; -- 未匹配到则插入新记录

关键配置说明 #

• KEYS：必须指定业务唯一键，确保合并时能精准匹配已有记录
• SEQUENCE BY：指定增量数据的处理顺序，避免因数据乱序导致的同步错误
• 触发模式：将DLT流水线配置为连续触发，实现对Bronze层新增数据的实时响应
• 数据质量校验：可在Silver层表中添加CONSTRAINT规则，过滤无效数据，比如：

  CREATE LIVE TABLE silver_table_y
  COMMENT "Silver层转换后数据存储表"
  CONSTRAINT valid_event_time CHECK (event_time > '2020-01-01')
  TBLPROPERTIES ("quality" = "silver")
  AS ...;
  

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