求助：Apache Beam/Dataflow CDC同步中BigQuery MERGE空值处理问题

针对你用Apache Beam/Dataflow同步MongoDB变更到BigQuery，需要MERGE日志表到目标表时处理「字段倒数第二条非空、最后一条为空」的场景，BigQuery不需要像其他数据库那样写自定义函数，用窗口函数预处理数据就能高效实现。

核心思路 #

在MERGE的USING子句中，先通过窗口函数LAG()获取每条最新记录的前一条（倒数第二条）字段值，把这些值和最新记录绑定，这样在UPDATE时就能直接引用前序值做逻辑判断，避免反复查询。

具体SQL实现 #

假设日志表为change_log，目标业务表为business_table，主键是user_id，业务字段包含name、email：

1. 预处理源数据（获取最新记录及前序字段值）

WITH enriched_source AS (
  SELECT
    user_id,
    name,
    email,
    operation_type,
    timestamp,
    -- 提取同一user_id下的上一条记录字段值
    LAG(name) OVER(PARTITION BY user_id ORDER BY timestamp DESC) AS prev_name,
    LAG(email) OVER(PARTITION BY user_id ORDER BY timestamp DESC) AS prev_email,
    -- 标记是否为最新变更记录
    ROW_NUMBER() OVER(PARTITION BY user_id ORDER BY timestamp DESC) AS rn
  FROM `your-project.your-dataset.change_log`
),
latest_changes AS (
  -- 只保留每个user_id的最新记录
  SELECT * FROM enriched_source WHERE rn = 1
)

2. 执行MERGE操作

MERGE INTO `your-project.your-dataset.business_table` trg
USING latest_changes src
ON trg.user_id = src.user_id

-- 匹配到现有记录时，根据业务逻辑更新字段
WHEN MATCHED THEN
  UPDATE SET
    name = CASE
      -- 按你的原逻辑：最新值非空则用最新值，否则设为null
      WHEN src.name IS NOT NULL THEN src.name
      ELSE NULL
      -- 如果需求是「最新值为空时保留前序非空值」，则改为 ELSE src.prev_name
    END,
    email = CASE
      WHEN src.email IS NOT NULL THEN src.email
      ELSE NULL
      -- ELSE src.prev_email
    END
-- 处理插入/替换操作（无匹配记录时新增）
WHEN NOT MATCHED AND src.operation_type IN ('插入', '替换') THEN
  INSERT (user_id, name, email)
  VALUES (src.user_id, src.name, src.email)
-- 处理删除操作（可选）
WHEN MATCHED AND src.operation_type = '删除' THEN
  DELETE;

关键优势 #

• 性能更优：LAG()窗口函数是批量计算，比自定义函数逐行查询的方式效率高很多，尤其适合大数据量的同步场景。
• 逻辑清晰：预处理阶段就把需要的前序字段值和最新记录绑定，MERGE的逻辑更直观，便于维护。

注意事项 #

• 确保timestamp字段的准确性，它是排序获取最新记录的核心依据；如果存在同一时间戳的多条变更，建议结合更精细的排序键（比如MongoDB的变更事件ID）。
• 根据实际业务需求调整CASE逻辑：比如如果空值代表「清除字段」，就直接用ELSE NULL；如果空值是误记录需要保留前序值，就用ELSE src.prev_xxx。

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