在BigQuery中获取每个ID值回落前的最高值

问题场景 #

现有数据表结构及示例数据如下：

ID | value | message_time
-------------------------
1  | 5     | 2023-07-01 10:00:00
1  | 7     | 2023-07-01 10:00:05
1  | 8     | 2023-07-01 10:00:12 <-- Highest Value
1  | 2     | 2023-07-01 10:00:14 <-- Dropped
1  | 4     | 2023-07-01 10:00:18
1  | 8     | 2023-07-01 10:00:28
1  | 12    | 2023-07-01 10:00:31 <-- Highest Value
1  | 5     | 2023-07-01 10:00:33 <-- Dropped
1  | 7     | 2023-07-01 10:00:38
1  | 9     | 2023-07-01 10:00:39
1  | 10    | 2023-07-01 10:00:40
1  | 14    | 2023-07-01 10:00:41 <-- Highest Value

需要提取每个ID在标记为Dropped的回落行之前的最高值记录，预期结果：

ID | value | message_time
-------------------------
1  | 8     | 2023-07-01 10:00:12
1  | 12    | 2023-07-01 10:00:31
1  | 14    | 2023-07-01 10:00:41

数据表包含10000+唯一ID，每个ID对应200+行数据，尝试过LAG()函数但无法满足需求，现提供解决方案。

解决方案 #

核心逻辑是按ID分组，将每个Dropped行及之前的未分组数据划分为独立区间，再在每个区间内筛选出value最大的记录。具体SQL如下：

WITH ranked_data AS (
  SELECT
    ID,
    value,
    message_time,
    -- 标记当前行是否为Dropped：这里假设回落判断为当前value小于前一行value
    -- 若表中有专门的Dropped标记列（如status='Dropped'），直接替换为该条件即可
    CASE WHEN value < LAG(value) OVER (PARTITION BY ID ORDER BY message_time) THEN 1 ELSE 0 END AS is_dropped,
    -- 累计生成区间分组：每遇到一个Dropped行，分组ID+1，划分出每个回落前的独立数据段
    SUM(CASE WHEN value < LAG(value) OVER (PARTITION BY ID ORDER BY message_time) THEN 1 ELSE 0 END) OVER (PARTITION BY ID ORDER BY message_time) AS interval_group
  FROM
    `your-project.your-dataset.your-table` -- 替换为你的表路径
),
max_per_interval AS (
  SELECT
    ID,
    interval_group,
    MAX(value) AS max_value,
    -- 若区间内有多个相同最大值，取最新的时间记录
    MAX(message_time) KEEP (DENSE_RANK LAST ORDER BY value) AS max_time
  FROM
    ranked_data
  GROUP BY
    ID, interval_group
)
SELECT
  r.ID,
  r.value,
  r.message_time
FROM
  ranked_data r
JOIN
  max_per_interval m
ON
  r.ID = m.ID
  AND r.interval_group = m.interval_group
  AND r.value = m.max_value
  AND r.message_time = m.max_time
ORDER BY
  r.ID, r.message_time;

代码说明 #

1. ranked_data CTE：

   • 用LAG()函数判断回落行，生成is_dropped标记（可根据实际业务规则调整判断条件）。
   • 通过SUM() OVER()累计回落标记值，生成interval_group，将每个回落点前后的数据划分为独立区间。

2. max_per_interval CTE：

   • 按ID和区间分组，计算每个区间的最大value，以及对应最新的时间戳（避免同一区间内多个相同最大值的冲突）。

3. 最终查询：

   • 关联原数据与区间最大值信息，筛选出每个区间的最高值记录。

性能说明 #

针对10000+ID、单ID200+行的数据量，该方案基于窗口函数和分组聚合实现，是BigQuery优化过的算子，性能表现稳定。

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