BigQuery增量更新累计列问题求助

计算规则梳理 #

• total_accumulated_margin：全量累计，新值 = 该用户历史累计值 + 当日daily_margin；如果是该用户首次数据，直接等于当日daily_margin。
• running_margin_30d：滚动30天求和，新值 = 该用户上一日的30天滚动值 + 当日daily_margin - 该用户30天前的daily_margin；如果30天前无数据，直接累加当前所有天数的和。

完整MERGE SQL示例 #

假设历史表为your_project.your_dataset.historical_margins，当日新增数据临时表为your_project.your_dataset.daily_new_data：

MERGE INTO `your_project.your_dataset.historical_margins` AS target
USING (
  SELECT
    nd.day,
    nd.person_id,
    nd.daily_margin,
    -- 获取该用户最新的历史累计值
    MAX(IF(h.day = (SELECT MAX(day) FROM `your_project.your_dataset.historical_margins` WHERE person_id = nd.person_id), h.total_accumulated_margin, NULL)) AS last_total,
    -- 获取该用户最新的30天滚动值
    MAX(IF(h.day = (SELECT MAX(day) FROM `your_project.your_dataset.historical_margins` WHERE person_id = nd.person_id), h.running_margin_30d, NULL)) AS last_running_30d,
    -- 获取30天前该用户的日边际值
    h_30d.daily_margin AS margin_30d_ago
  FROM `your_project.your_dataset.daily_new_data` AS nd
  LEFT JOIN `your_project.your_dataset.historical_margins` AS h
    ON nd.person_id = h.person_id
  LEFT JOIN `your_project.your_dataset.historical_margins` AS h_30d
    ON nd.person_id = h_30d.person_id
    AND DATE_SUB(nd.day, INTERVAL 30 DAY) = h_30d.day
  GROUP BY nd.day, nd.person_id, nd.daily_margin, h_30d.daily_margin
) AS source
ON target.day = source.day AND target.person_id = source.person_id
WHEN NOT MATCHED THEN
  INSERT (day, person_id, daily_margin, running_margin_30d, total_accumulated_margin)
  VALUES (
    source.day,
    source.person_id,
    source.daily_margin,
    COALESCE(source.last_running_30d, 0) + source.daily_margin - COALESCE(source.margin_30d_ago, 0),
    COALESCE(source.last_total, 0) + source.daily_margin
  )
WHEN MATCHED THEN
  UPDATE SET
    daily_margin = source.daily_margin,
    running_margin_30d = source.last_running_30d + source.daily_margin - COALESCE(source.margin_30d_ago, 0),
    total_accumulated_margin = source.last_total + source.daily_margin;

关键细节说明 #

1. 获取最新历史值：通过子查询SELECT MAX(day)...定位该用户的最新历史记录，确保拿到正确的上一日累计值。
2. 空值处理：用COALESCE把空值替换为0，避免首次插入或30天无数据时出现计算错误。
3. 匹配逻辑：通过day和person_id组合键匹配，防止重复插入或错误更新已有记录。

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