PySpark技术实现：计算聊天室从断连到重连的平均耗时

嘿，这个需求我之前做过类似的统计，咱们一步步拆解来解决它！首先得明确核心要求：每个聊天室，每次从第一次进入断连状态（isConnected=false）到恢复连接（isConnected=true）的时长，然后求这些时长的平均值。

先对照数据理清楚例子：

• roomId=1：连续断连的记录是3000和4000，首次断连是3000，之后第一次重连是8000，耗时5000，所以平均就是5000。
• roomId=3：有两次独立断连阶段：1000→2000（耗时1000）、5000→7000（耗时2000），平均就是(1000+2000)/2=1500，完全匹配预期输出。

接下来用SQL实现这个逻辑，这里用窗口函数来处理，简洁高效，适合MySQL 8.0+、PostgreSQL等支持窗口函数的数据库：

WITH status_with_next_true AS (
    SELECT 
        roomId,
        timeStamp AS disconnect_time,
        -- 为每个断连记录找到同房间后续第一个重连的时间
        LEAD(CASE WHEN isConnected THEN timeStamp END) OVER (
            PARTITION BY roomId 
            ORDER BY timeStamp
        ) AS reconnect_time
    FROM chat_room_status
    WHERE isConnected = false
),
valid_reconnect_periods AS (
    SELECT 
        roomId,
        reconnect_time - disconnect_time AS reconnect_duration
    FROM status_with_next_true
    WHERE reconnect_time IS NOT NULL -- 过滤掉没有后续重连的断连记录
    -- 只保留每个断连阶段的第一条记录（首次断连时间）
    AND disconnect_time = (
        SELECT MIN(s2.timeStamp)
        FROM chat_room_status s2
        WHERE s2.roomId = status_with_next_true.roomId
        AND s2.timeStamp >= disconnect_time
        AND s2.isConnected = false
        -- 判断当前记录是断连阶段的起点：上一条记录不是断连状态，或者是第一条记录
        AND COALESCE(
            (SELECT s3.isConnected FROM chat_room_status s3 
             WHERE s3.roomId = s2.roomId AND s3.timeStamp < s2.timeStamp 
             ORDER BY s3.timeStamp DESC LIMIT 1),
            true
        ) = true
    )
)
SELECT 
    roomId,
    AVG(reconnect_duration) AS avgConTime
FROM valid_reconnect_periods
GROUP BY roomId
ORDER BY roomId;

代码逻辑拆解： #

1. 第一个CTE status_with_next_true：筛选所有断连记录，并用LEAD窗口函数找到每个断连记录之后同房间的第一个重连时间。
2. 第二个CTE valid_reconnect_periods：
   • 过滤掉没有后续重连的断连记录（比如如果某个房间最后一条是断连且没有重连，就不统计）。
   • 只保留每个连续断连阶段的第一条记录（也就是首次断连的时间），避免重复计算同一个断连阶段的时长。
3. 最终查询：按房间分组，计算所有有效重连时长的平均值，得到预期结果。

如果你的数据库不支持窗口函数（比如MySQL 5.x），可以用自连接的方式实现，但窗口函数的写法更易读和维护。

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