Impala SQL按30分钟间隔重置规则计算同ID时间戳序列计数

现有一张包含ID、Datetime两个字段的表，原始样例数据如下：

ID          Datetime            
123         12Sep2021 10:00       
123         12Sep2021 10:10
123         12Sep2021 10:25
123         12Sep2021 10:40
123         12Sep2021 10:52
123         12Sep2021 11:20
456         01Oct2021 09:00
456         01Oct2021 09:10
456         01Oct2021 09:40

需求为新增Count字段，规则如下：同ID下第一条记录Count为1；后续记录与当前组起始时间间隔小于30分钟则Count持续递增；若间隔大于等于30分钟则Count重置为1，后续时间差以重置记录的时间为新的组起始时间计算。预期输出如下：

ID          Datetime              Count      
123         12Sep2021 10:00        1
123         12Sep2021 10:10        2
123         12Sep2021 10:25        3
123         12Sep2021 10:40        1
123         12Sep2021 10:52        2 
123         12Sep2021 11:20        1
456         01Oct2021 09:00        1
456         01Oct2021 09:10        2
456         01Oct2021 09:40        1

原Python Pandas实现方案在数据量过大时存在单节点内存不足问题，以下为Impala SQL无需循环的实现方案：

方案1：MATCH_RECOGNIZE（优先推荐，Impala 3.2及以上版本支持） #

该方案使用Impala内置的模式匹配语法，分布式执行效率最高，代码如下：

SELECT ID, Datetime, Count
FROM 你的表名
MATCH_RECOGNIZE (
    PARTITION BY ID -- 按ID分组
    ORDER BY Datetime -- 组内按时间排序
    MEASURES COUNT(*) AS Count -- 每个匹配组内的行号即为Count值
    ONE ROW PER MATCH -- 每行返回一条结果
    AFTER MATCH SKIP PAST LAST ROW -- 匹配完一组后从下一行开始新的匹配
    PATTERN (a b*) -- 匹配规则：1个起始行a，后跟任意个符合条件的行b
    DEFINE b AS UNIX_TIMESTAMP(Datetime) - UNIX_TIMESTAMP(FIRST(a.Datetime)) < 30*60 -- b的判定规则：和当前组起始行a的时间差小于30分钟
)
ORDER BY ID, Datetime;

方案2：递归CTE（Impala 2.11及以上版本支持） #

该方案逻辑和你原有Pandas代码完全一致，容易理解，代码如下：

WITH numbered AS (
    -- 第一步：给同ID的所有记录按时间排序加行号
    SELECT 
        ID,
        Datetime,
        ROW_NUMBER() OVER(PARTITION BY ID ORDER BY Datetime) AS rn
    FROM 你的表名
),
recursive_cte AS (
    -- 锚点：每个ID的第一条记录，Count固定为1，组起始时间为当前时间
    SELECT 
        ID,
        Datetime,
        rn,
        Datetime AS group_start,
        1 AS Count
    FROM numbered
    WHERE rn = 1
    UNION ALL
    -- 递归处理后续每条记录
    SELECT 
        n.ID,
        n.Datetime,
        n.rn,
        CASE WHEN UNIX_TIMESTAMP(n.Datetime) - UNIX_TIMESTAMP(r.group_start) >= 1800 THEN n.Datetime ELSE r.group_start END AS group_start,
        CASE WHEN UNIX_TIMESTAMP(n.Datetime) - UNIX_TIMESTAMP(r.group_start) >= 1800 THEN 1 ELSE r.Count + 1 END AS Count
    FROM numbered n
    INNER JOIN recursive_cte r ON n.ID = r.ID AND n.rn = r.rn + 1
)
SELECT ID, Datetime, Count
FROM recursive_cte
ORDER BY ID, rn;

两种方案都不需要将全量数据加载到单节点内存，依托Impala的分布式计算能力可支持TB级数据量的计算。

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