如何从带状态与时间的事件流中高效计算状态持续时间？

场景与需求 #

按按钮ID分组处理状态事件流，计算从第一个ON状态到下一个OFF状态的持续时间，输出包含按钮ID、ON状态起始时间、持续时长的结果流。

原事件流定义 #

CREATE STREAM events (id STRING KEY, state STRING, time TIMESTAMP) WITH (KAFKA_TOPIC='events', value_format='avro', partitions=3)

测试输入数据 #

INSERT INTO events (id, state, time) VALUES ('BUTTON-1', 'OFF', '2023-06-19T20:00:00Z')
INSERT INTO events (id, state, time) VALUES ('BUTTON-1', 'OFF', '2023-06-19T20:01:00Z')
INSERT INTO events (id, state, time) VALUES ('BUTTON-1', 'ON', '2023-06-19T20:02:00Z')
INSERT INTO events (id, state, time) VALUES ('BUTTON-1', 'ON', '2023-06-19T20:03:00Z')
INSERT INTO events (id, state, time) VALUES ('BUTTON-1', 'OFF', '2023-06-19T20:04:00Z')
INSERT INTO events (id, state, time) VALUES ('BUTTON-1', 'ON', '2023-06-19T20:05:00Z')
INSERT INTO events (id, state, time) VALUES ('BUTTON-1', 'OFF', '2023-06-19T20:06:00Z')

期望输出 #

原方案问题 #

原方案通过分流ON/OFF事件再关联分组实现，但多次分组和关联导致效率低下：

-- 1. 分流ON/OFF事件
CREATE STREAM on_events AS SELECT * FROM events WHERE state = 'ON';
CREATE STREAM off_events AS SELECT * FROM events WHERE state = 'OFF';

-- 2. 关联分流并生成表
CREATE TABLE on_next_off AS
    SELECT on_events.id AS id, on_events.time AS on_time, min(off_events.time) as off_time
    FROM on_events
    INNER JOIN off_events ON on_events.id = off_events.id
    WHERE on_events.time < off_events.time
    GROUP BY on_events.id, on_events.time

-- 3. 计算持续时间
CREATE TABLE on_duration AS
    SELECT id, on_time, off_time - ARRAY_MIN(COLLECT_LIST(on_time)) as duration
    FROM on_next_off
    GROUP BY id, on_time

高效实现方案（无需UDF） #

利用KSQL窗口函数和状态追踪能力，通过一次流处理完成计算，避免多次关联和分组：

方法1：基于状态表追踪最新ON时间 #

-- 创建状态表，存储每个ID的最新ON起始时间
CREATE TABLE latest_on_time AS
SELECT
    id,
    LATEST_BY_OFFSET(time) AS on_start_time
FROM events
WHERE state = 'ON'
GROUP BY id;

-- 关联事件流与状态表，仅在OFF事件触发时计算持续时间
CREATE STREAM button_on_duration AS
SELECT
    e.id,
    lot.on_start_time AS on_time,
    (UNIX_TIMESTAMP(e.time) - UNIX_TIMESTAMP(lot.on_start_time)) * 1000 AS duration
FROM events e
INNER JOIN latest_on_time lot ON e.id = lot.id
WHERE e.state = 'OFF'
  AND lot.on_start_time IS NOT NULL
  -- 仅处理从ON切换到OFF的事件，过滤重复OFF
  AND LAG(e.state) OVER (PARTITION BY e.id ORDER BY e.time) = 'ON';

方法2：基于窗口函数识别状态周期 #

-- 标记状态周期的起始点（状态切换时）
CREATE STREAM state_periods AS
SELECT
    id,
    state,
    time,
    CASE WHEN state != LAG(state) OVER (PARTITION BY id ORDER BY time) THEN time ELSE NULL END AS period_start
FROM events;

-- 向前填充周期起始时间，为每个事件绑定所属周期的起始
CREATE STREAM state_periods_with_start AS
SELECT
    id,
    state,
    time,
    LAST_VALUE(period_start IGNORE NULLS) OVER (PARTITION BY id ORDER BY time ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS period_start
FROM state_periods;

-- 提取OFF事件，计算对应ON周期的持续时间
CREATE STREAM button_on_duration_final AS
SELECT
    id,
    period_start AS on_time,
    (UNIX_TIMESTAMP(time) - UNIX_TIMESTAMP(period_start)) * 1000 AS duration
FROM state_periods_with_start
WHERE state = 'OFF'
  AND period_start IS NOT NULL
  -- 确保周期起始为ON状态，过滤初始OFF周期
  AND (SELECT state FROM events WHERE id = state_periods_with_start.id AND time = period_start) = 'ON';

上述方案通过单次流处理和轻量状态追踪，大幅提升处理效率，且无需自定义UDF。

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