在BigQuery窗口函数中使用QUALIFY实现精准任务金额求和

背景与需求 #

我有一个任务事件数据集，每条记录对应一个任务，每个任务关联一位用户，字段包括唯一id、金额amt、创建日期date。当任务金额更新时，会新增一条记录，包含对应的最后更新时间last_update_Date和版本号version。

需求：对每个任务，计算该用户所有其他任务中，最后更新时间早于当前任务创建日期的最新版本金额总和。

举个例子：在提供的示例数据中，id为3的任务创建日期是2020-02-11，需取用户a的其他任务（id=1和id=2）中，last_update_Date早于该日期的最新版本——id=1的version 1（金额20）、id=2的version 0（金额5），总和为25。

尝试的问题代码 #

我曾用窗口函数实现，但代码会累加任务的所有事件记录，不符合需求：

select task_id, SUM( task_amt) over(partition by user  order by DATE_DIFF(TIMESTAMP(date), TIMESTAMP('2007-12-31 12:00:00+00'), SECOND) 
 range between 7776000 preceding and 1 preceding)) as task_amt from table

疑问 #

能否通过在窗口函数中添加QUALIFY子句实现需求？还是需要Spark等其他工具？

数据集（BigQuery建表&插入语句） #

CREATE TABLE table (
  id INT64,
  user STRING,
  amt FLOAT64,
  date DATE,
  last_update_Date TIMESTAMP,
  version INT64
);

INSERT INTO table (id, user, amt, date, last_update_Date, version)
VALUES
  (1, 'a', 10.0, '2020-01-01', '2020-01-01 12:00:00', 0),
  (1, 'a', 20.0, '2020-01-01', '2020-02-01 12:00:00', 1),
  (1, 'a', 30.0, '2020-01-01', '2020-03-01 12:00:00', 2),
  (2, 'a', 5.0, '2020-02-10', '2020-02-10 12:00:00', 0),
  (2, 'a', 10.0, '2020-02-10', '2020-03-10 12:00:00', 1);

INSERT INTO table (id, user, amt, date, last_update_Date, version)
VALUES
  (3, 'a', 100.0, '2020-02-11', '2020-02-11 12:00:00', 0),
  (3, 'a', 200.0, '2020-02-11', '2020-02-15 12:00:00', 1);

解决方案（BigQuery实现，无需Spark） #

完全可以通过QUALIFY子句在BigQuery中实现需求，核心思路是先筛选出每个任务在目标时间点前的最新版本，再关联求和。

实现代码 #

WITH task_info AS (
  -- 提取每个任务的唯一创建日期（同一任务id的date固定）
  SELECT DISTINCT
    id AS target_task_id,
    user,
    date AS target_create_date
  FROM `table`
)
SELECT
  ti.target_task_id,
  SUM(t.amt) AS sum_qualified_amt
FROM task_info ti
LEFT JOIN (
  -- 用QUALIFY筛选每个任务id的最新版本
  SELECT
    id,
    user,
    amt,
    last_update_Date,
    QUALIFY ROW_NUMBER() OVER (PARTITION BY user, id ORDER BY last_update_Date DESC) = 1
  FROM `table`
) t
  ON ti.user = t.user
  AND t.id != ti.target_task_id -- 排除当前任务
  AND t.last_update_Date < TIMESTAMP(ti.target_create_date) -- 最后更新时间早于当前任务创建日期
GROUP BY ti.target_task_id
ORDER BY ti.target_task_id;

代码说明 #

1. task_info CTE：提取每个任务的唯一id、所属用户和创建日期，避免重复计算同一任务的多个版本。
2. 子查询中用QUALIFY+ROW_NUMBER()：按任务id分组，按last_update_Date降序排序，筛选出每个任务的最新版本。
3. 关联与求和：将目标任务与筛选后的其他任务最新版本关联，排除自身任务，筛选时间条件后求和。

执行上述代码后，id=3的任务会得到预期结果25，其他任务的计算也符合需求。

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