如何在SQL/Snowflake中提取时序数据中超阈值时段的起止时间

原始时序数据 #

INSERT INTO timeseries (timestamp, value)
VALUES
  ('2022-01-01 00:00:00', 0.89),
  ('2022-01-01 10:01:00', 6.89),
  ('2022-01-02 10:01:21', 10.99),
  ('2022-01-02 10:07:00', 11.89),
  ('2022-01-02 12:01:00', 0.89),
  ('2022-01-02 13:07:00', 6.39),
  ('2022-01-02 14:00:00', 0.69),
  ('2022-01-03 14:02:00', 5.39),
  ('2022-01-03 15:04:00', 6.89),
  ('2022-01-03 15:00:00', 7.3),
  ('2022-01-03 15:10:00', 1.89),
  ('2022-01-03 15:50:00', 0.8);

需求说明 #

需要提取value > 5的连续时段的最小、最大timestamp，并计算每个时段的分钟差值。已知符合条件的3个时段：

min                   max
2022-01-01 10:01:00   2022-01-02 10:07:00
2022-01-02 13:07:00   2022-01-02 13:07:00
2022-01-03 14:02:00   2022-01-03 15:00:00

尝试过的代码（未实现分组） #

WITH CTE AS (
SELECT CASE WHEN VALUE>5 THEN 'ON' ELSE 'OFF' END STATUS , TIMESTAMP, VALUE
FROM TIMESERIES)
SELECT ROW_NUMBER() OVER(PARTITION BY STATUS ORDER BY TIMESTAMP) RN,TIMESTAMP,VALUE FROM CTE
ORDER BY TIMESTAMP;

解决方案：间隙与孤岛问题解法 #

核心思路是通过行号差值法给连续的超阈值记录分配相同分组ID，实现按连续时段分组：

1. 标记每条记录是否满足value > 5的条件
2. 生成全局行号（按timestamp排序）和分区行号（按是否超阈值分组排序）
3. 用全局行号减去分区行号，差值相同的记录属于同一连续时段
4. 按分组ID聚合，提取时段的起止时间并计算时长

兼容Snowflake的SQL代码

WITH tagged_data AS (
    -- 标记是否超阈值
    SELECT 
        timestamp,
        value,
        CASE WHEN value > 5 THEN 1 ELSE 0 END AS is_over_threshold
    FROM timeseries
),
row_number_gen AS (
    -- 生成全局行号和分区行号
    SELECT 
        timestamp,
        value,
        is_over_threshold,
        ROW_NUMBER() OVER(ORDER BY timestamp) AS global_row,
        ROW_NUMBER() OVER(PARTITION BY is_over_threshold ORDER BY timestamp) AS threshold_row
    FROM tagged_data
),
grouped_records AS (
    -- 计算分组ID，连续超阈值记录的group_id相同
    SELECT 
        timestamp,
        is_over_threshold,
        global_row - threshold_row AS group_id
    FROM row_number_gen
    WHERE is_over_threshold = 1 -- 只保留超阈值的记录
)
-- 聚合得到每个连续时段的起止时间和时长
SELECT 
    MIN(timestamp) AS min_timestamp,
    MAX(timestamp) AS max_timestamp,
    TIMESTAMPDIFF(MINUTE, MIN(timestamp), MAX(timestamp)) AS duration_minutes
FROM grouped_records
GROUP BY group_id
ORDER BY min_timestamp;

代码说明

• tagged_data：给每条记录打标，区分是否超阈值
• row_number_gen：生成全局排序的行号，以及按阈值状态分区的行号。对于连续的超阈值记录，全局行号和分区行号的增长同步，差值保持不变
• grouped_records：筛选超阈值记录，通过行号差得到分组ID，同一连续时段的记录会被分到同一组
• 最后一步聚合分组，得到每个时段的起止时间和分钟级时长

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