基于动态时间戳与用户列新增状态列的SQL/PySpark实现求助

问题说明 #

原始数据 #

id, name, timestamp
1, David, 2022/01/01 10:00
2, David, 2022/01/01 10:30
3, Diego, 2022/01/01 10:59
4, David, 2022/01/01 10:59
5, David, 2022/01/01 11:01
6, Diego, 2022/01/01 12:00
7, David, 2022/01/01 12:00
8, David, 2022/01/01 12:05
9, Diego, 2022/01/01 12:30

业务逻辑 #

David和Diego参与按键游戏：

• 首次按键后1小时内的后续按键属于同一游戏周期
• 超出1小时后再次按键则视为开启新周期
• 新增status列：周期内首次按键标记为0（start），其余标记为1（playing）

分步实现方案 #

步骤1：创建临时视图并转换时间格式 #

先将原始数据加载为临时视图，同时把字符串类型的timestamp转换为Spark可识别的时间类型，方便后续时间计算。

-- 创建原始数据临时视图
CREATE OR REPLACE TEMP VIEW game_events AS
SELECT 
    id,
    name,
    to_timestamp(timestamp, 'yyyy/MM/dd HH:mm') AS event_time
FROM (
    VALUES
        (1, 'David', '2022/01/01 10:00'),
        (2, 'David', '2022/01/01 10:30'),
        (3, 'Diego', '2022/01/01 10:59'),
        (4, 'David', '2022/01/01 10:59'),
        (5, 'David', '2022/01/01 11:01'),
        (6, 'Diego', '2022/01/01 12:00'),
        (7, 'David', '2022/01/01 12:00'),
        (8, 'David', '2022/01/01 12:05'),
        (9, 'Diego', '2022/01/01 12:30')
) AS t(id, name, timestamp);

步骤2：计算每个事件的周期起始时间 #

通过窗口函数动态判断每个事件所属的周期起始时间：如果是用户的首个事件，或者当前事件距离上一周期起始时间超过1小时，则当前事件作为新周期的起始；否则沿用之前的周期起始时间。

-- 计算每个事件所属的周期起始时间，生成临时视图
CREATE OR REPLACE TEMP VIEW game_events_with_cycle AS
SELECT 
    id,
    name,
    event_time,
    -- 动态确定周期起始时间
    last(
        when(
            row_number() OVER (PARTITION BY name ORDER BY event_time) = 1
            OR event_time > date_add(lag(cycle_start) OVER (PARTITION BY name ORDER BY event_time), 1/24)
            , event_time
        )
        , True
    ) OVER (PARTITION BY name ORDER BY event_time ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS cycle_start
FROM (
    -- 初始给每个事件设置临时起始时间，后续通过窗口函数修正
    SELECT 
        id,
        name,
        event_time,
        event_time AS cycle_start
    FROM game_events
) t;

步骤3：生成status列 #

按用户和周期起始时间分组，判断每个事件是否为该周期的首个事件，生成对应的status值，并将时间格式转回原始字符串格式。

-- 最终生成带status的结果
SELECT 
    id,
    name,
    date_format(event_time, 'yyyy/MM/dd HH:mm') AS timestamp,
    CASE 
        WHEN row_number() OVER (PARTITION BY name, cycle_start ORDER BY event_time) = 1 THEN 0
        ELSE 1
    END AS status
FROM game_events_with_cycle
ORDER BY id;

结果验证 #

执行上述SQL后，将得到符合需求的结果：

id, name, timestamp, status
1, David, 2022/01/01 10:00, 0  <--- David starts playing
2, David, 2022/01/01 10:30, 1  <--- David keeps playing the game that he started at the id 1
3, Diego, 2022/01/01 10:59, 0  <--- Diego starts playing
4, David, 2022/01/01 10:59, 1  <--- David keeps playing the game that he started at the id 1
5, David, 2022/01/01 11:01, 0  <--- David starts playing again
6, Diego, 2022/01/01 12:00, 0  <--- Diego starts playing again
7, David, 2022/01/01 12:00, 1  <--- David keeps playing the game that he started at the id 5
8, David, 2022/01/01 12:05, 0  <--- David start playing again
9, Diego, 2022/01/01 12:30, 1  <--- Diego keeps playing the game that he started at the id 6

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