Databricks用ANSI SQL复刻CONDITIONAL_TRUE_EVENT实现事件分组

逻辑说明 #

从Snowflake迁移脚本到Databricks时，可通过标准ANSI SQL窗口函数组合替代专有CONDITIONAL_TRUE_EVENT函数，实现事件分组逻辑：

• 分组维度：user_id + device_id
• 分组规则：同维度下按事件时间排序，相邻事件时间间隔≤300秒（5分钟）归为同一事件组；间隔超阈值则生成新组
• 组号规则：每个事件组按其首次出现的时间顺序分配全局唯一的连续ID，同组事件即使中间穿插其他维度事件，只要时间间隔符合阈值仍保留原组ID，完全匹配样例预期。

测试表与样例数据 #

CREATE TABLE events
     (
        event_timestamp timestamp,
        user_id bigint,
        device_id bigint
     );

INSERT INTO events VALUES
  ('2022-07-12 05:00:00',1,1),
  ('2022-07-12 05:03:00',1,1),
  ('2022-07-12 05:04:00',1,2),
  ('2022-07-12 05:05:00',1,2),
  ('2022-07-12 05:06:00',2,1),
  ('2022-07-12 05:07:00',1,1),
  ('2022-07-12 05:15:00',1,1);

预期输出结果 #

实现SQL（兼容Databricks，ANSI标准语法） #

核心逻辑分三步：

1. 用LAG函数取同维度下上一个事件的发生时间
2. 同维度内判断时间间隔，生成本地会话标记，累加得到维度内会话ID（这一步就是CONDITIONAL_TRUE_EVENT的等效实现）
3. 取每个会话的首次出现时间，用DENSE_RANK生成全局连续的组ID

WITH step1 AS (
  SELECT
    event_timestamp,
    user_id,
    device_id,
    -- 取同用户、同设备维度下的上一个事件时间
    LAG(event_timestamp) OVER (
      PARTITION BY user_id, device_id 
      ORDER BY event_timestamp
    ) AS last_same_dim_ts
  FROM events
),
step2 AS (
  SELECT
    event_timestamp,
    user_id,
    device_id,
    -- 间隔超300秒/无上一个事件则标记为新会话起点，累加生成维度内会话ID
    SUM(
      CASE
        WHEN last_same_dim_ts IS NULL 
          OR TIMESTAMPDIFF(SECOND, last_same_dim_ts, event_timestamp) > 300
        THEN 1 ELSE 0
      END
    ) OVER (
      PARTITION BY user_id, device_id 
      ORDER BY event_timestamp
    ) AS local_session_id
  FROM step1
),
step3 AS (
  SELECT
    event_timestamp,
    user_id,
    device_id,
    -- 取每个会话的首次出现时间，用于分配全局组号
    MIN(event_timestamp) OVER (
      PARTITION BY user_id, device_id, local_session_id
    ) AS session_first_ts
  FROM step2
)
SELECT
  event_timestamp,
  user_id,
  device_id,
  -- 按会话首次出现顺序分配全局连续组号
  DENSE_RANK() OVER (ORDER BY session_first_ts) AS group_id
FROM step3
ORDER BY event_timestamp;

注意事项 #

• 不要直接全局排序后累加新会话标记，该写法会导致中间穿插其他维度事件后，回到历史同维度会话时组号被错误抬高，无法匹配预期结果
• 上述语法使用的LAG、SUM窗口、DENSE_RANK、TIMESTAMPDIFF均为ANSI SQL标准函数，在Databricks、Spark SQL、Presto等主流引擎中均可直接运行

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