时间重叠事件（重叠区间）识别分组的高效算法

这个问题本质是时间区间连通分量标记问题：只要事件间存在直接/间接的时间重叠（含传递性重叠、端点接触重叠、瞬时点事件重叠），就归为同一分组，工业界大规模数据集下的最优通用方案时间复杂度为O(n log n)，远优于基础两两比对的O(n²)方案，具体实现逻辑如下。

核心算法逻辑 #

采用「端点排序+线性扫描+活跃集合维护」的扫描线思路，不需要额外复杂数据结构，同时天然支持瞬时点事件、端点接触重叠的判定：

• 第一步：端点拆分
  把每个事件拆成两个时间点，绑定事件ID和点类型：
  • 起点：类型标记为start，代表事件进入生效状态
  • 终点：类型标记为end，代表事件退出生效状态
• 第二步：端点排序
  所有拆分出的时间点按两个规则升序排列：
  1. 时间值更小的点排在前面
  2. 时间值相同的点，start类型排在end类型前面

  这个排序规则刚好适配示例中的判定逻辑：同时间点先处理新事件进入，再处理旧事件退出，既可以正确识别「前一个事件的结束时间等于后一个事件的开始时间」的端点重叠（比如示例中E的结束时间和A的开始时间同为15:00:01，会被判定为同组），也能正确处理起止时间相同的瞬时点事件（比如示例中的D事件，同时间的start和end会按先入后出的顺序处理，不会漏判重叠）。如果业务要求端点接触不算重叠，只需要把同时间点的排序改成end在前、start在后即可，不需要调整其他逻辑。

• 第三步：线性扫描分组
  扫描排序后的时间点，全程维护两个状态：
  • active_set：当前时间点下所有处于生效状态（已触发起点、未触发终点）的事件集合
  • current_group_id：当前连通分组的ID
    扫描时按点类型处理：
  • 遇到start点：如果当前active_set为空，说明进入了一个新的独立连通区间，生成新的分组ID；把当前事件加入active_set，标记该事件的分组为current_group_id
  • 遇到end点：把对应事件从active_set中移除即可

用题目给出的示例数据验证：拆分排序后的点扫描完成后，B/C/D归为组1，A/E归为组2，F单独归为组3，和预期结果完全一致。

可运行代码示例（Python） #

def group_overlapping_events(events):
    """
    :param events: 事件列表，格式为 [(事件ID, 开始时间, 结束时间)]，时间支持可比较的字符串、时间戳、datetime类型
    :return: 分组映射字典，key为事件ID，value为对应分组ID
    """
    points = []
    for event_id, start, end in events:
        # 类型码0代表start，1代表end，同时间下0自然排1前面，满足排序规则
        points.append((start, 0, event_id))
        points.append((end, 1, event_id))
    # 按规则排序端点
    points.sort()
    
    active_set = set()
    group_map = {}
    current_gid = 0
    
    for _, point_type, event_id in points:
        if point_type == 0:
            # 处理起点
            if not active_set:
                current_gid += 1
            group_map[event_id] = current_gid
            active_set.add(event_id)
        else:
            # 处理终点
            active_set.remove(event_id)
    return group_map

# 测试题目给出的示例数据集
if __name__ == "__main__":
    test_events = [
        ("A", "15:00:01", "15:01:03"),
        ("B", "10:12:11", "10:15:09"),
        ("C", "10:13:59", "10:14:33"),
        ("D", "10:12:11", "10:12:11"),
        ("E", "14:59:30", "15:00:01"),
        ("F", "17:00:12", "17:01:01")
    ]
    result = group_overlapping_events(test_events)
    print(result)
    # 输出: {'A': 2, 'B': 1, 'C': 1, 'D': 1, 'E': 2, 'F': 3}，完全符合预期

性能与扩展说明 #

• 常规单机场景下，百万级事件量可以在百毫秒级完成计算，千万级事件配合外排逻辑也可以轻松处理
• 分布式场景下可以按时间分片处理，只需要在分片边界维护跨片的活跃事件集合，就能实现多节点并行计算，逻辑不需要做大幅修改
• 空间复杂度为O(n)，主要用于存储拆分的端点和活跃事件集合，无额外内存开销

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