在Amazon Redshift中计算用户最长B类连续记录及末尾连续长度

问题背景 #

现有如下Redshift表结构及数据：

create table streak(id int, "user" int, category char);

insert into streak values(1, 1, 'A');
insert into streak values(2, 1, 'B');
insert into streak values(3, 1, 'A');
insert into streak values(4, 1, 'B');
insert into streak values(5, 1, 'B');

insert into streak values(6, 2, 'A');
insert into streak values(7, 2, 'A');
insert into streak values(8, 2, 'B');
insert into streak values(9, 2, 'B');
insert into streak values(10,2, 'B');

insert into streak values(11,3, 'B');
insert into streak values(12,3, 'A');

需要为每个用户计算两个指标：

• B类记录的最长连续长度
• 截至最后一条记录的B类连续长度

预期结果：

user|longest_streak|last_streak
1|2|2
2|3|3
3|1|0

解决方案 #

可以通过窗口函数识别连续的B类记录段，再进行聚合计算，以下是实现SQL：

WITH streak_groups AS (
    SELECT
        "user",
        category,
        id,
        -- 标记连续B段的起始点：当前是B且上一条不是B时生成1，否则0
        SUM(CASE WHEN category = 'B' AND LAG(category) OVER (PARTITION BY "user" ORDER BY id) != 'B' THEN 1
            WHEN category != 'B' THEN 0
            ELSE 0 END) OVER (PARTITION BY "user" ORDER BY id) AS group_id
    FROM streak
),
streak_lengths AS (
    SELECT
        "user",
        group_id,
        COUNT(*) AS streak_len,
        -- 标记该组是否是用户的最后一组
        MAX(id) OVER (PARTITION BY "user") = MAX(id) OVER (PARTITION BY "user", group_id) AS is_last_group
    FROM streak_groups
    WHERE category = 'B'
    GROUP BY "user", group_id
),
user_max_streak AS (
    SELECT
        "user",
        COALESCE(MAX(streak_len), 0) AS longest_streak
    FROM streak_lengths
    GROUP BY "user"
    -- 补充无B记录的用户
    UNION ALL
    SELECT
        "user",
        0 AS longest_streak
    FROM streak
    WHERE "user" NOT IN (SELECT "user" FROM streak_lengths)
    GROUP BY "user"
),
user_last_streak AS (
    SELECT
        "user",
        CASE WHEN is_last_group THEN streak_len ELSE 0 END AS last_streak
    FROM streak_lengths
    WHERE is_last_group
    UNION ALL
    SELECT
        "user",
        0 AS last_streak
    FROM streak
    WHERE "user" NOT IN (SELECT "user" FROM streak_lengths WHERE is_last_group)
    GROUP BY "user"
)
SELECT
    ums."user",
    ums.longest_streak,
    uls.last_streak
FROM user_max_streak ums
JOIN user_last_streak uls ON ums."user" = uls."user"
ORDER BY ums."user";

逻辑说明 #

1. streak_groups CTE：通过LAG()函数对比当前与上一条记录的category，标记连续B段的起始点，并用SUM()累加生成每个连续B段的唯一group_id。
2. streak_lengths CTE：筛选B类记录，按用户和group_id分组计算连续段长度，同时标记该段是否为用户的最后一个记录段。
3. user_max_streak CTE：按用户聚合得到最长B类连续长度，用COALESCE和UNION ALL处理无B记录的用户（默认0）。
4. user_last_streak CTE：提取用户最后一个B类连续段的长度，无有效末尾B段则返回0，同时补充无B记录的用户。
5. 最后关联两个结果集，得到最终指标。

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