基于分段ID将重叠区间集合划分为非重叠区间的技术需求

嘿，这个场景我之前在处理时序分段数据时遇到过，给你一套落地性强的实现方案，不管你用Python、SQL还是其他语言，核心逻辑都是通的：

咱们的目标是把同一ID下的重叠区间拆成完全无重叠的子段，核心步骤分这几步：

• 先分组：把数据集按segment_id拆成独立的组，只处理同ID内的区间重叠，不同ID之间互不干扰
• 排序区间：对每个ID下的所有区间，按start升序排序；如果start相同，就按end升序排，保证后续处理的顺序一致性
• 提取关键分割点：收集当前ID下所有区间的start和end值，去重后按升序排列，这些点就是拆分无重叠子段的边界
• 生成无重叠子段：遍历相邻的分割点，把[分割点i, 分割点i+1]作为一个子区间，这些子段天然是无重叠的
• （可选）关联负载数据：如果需要每个子段的流量负载，就遍历每个子段，找到所有包含它的原区间，按需求做聚合（求和、取最大/平均都可以）

假设你的数据是列表字典格式，直接用这个函数就能处理：

data = [
    {"segment_id": "A", "start": 10, "end": 30, "length": 20, "load": 5},
    {"segment_id": "A", "start": 20, "end": 40, "length": 20, "load": 3},
    {"segment_id": "B", "start": 5, "end": 15, "length": 10, "load": 2},
    {"segment_id": "B", "start": 10, "end": 20, "length": 10, "load": 4},
]

def split_overlapping_segments(data):
    from collections import defaultdict
    
    # 1. 按segment_id分组，把同ID的区间放一起
    grouped_data = defaultdict(list)
    for item in data:
        grouped_data[item["segment_id"]].append(item)
    
    final_result = []
    
    for seg_id, segments in grouped_data.items():
        # 2. 按start升序排序，确保处理顺序正确
        sorted_segs = sorted(segments, key=lambda x: (x["start"], x["end"]))
        
        # 3. 收集所有端点，去重后排序
        boundary_points = set()
        for seg in sorted_segs:
            boundary_points.add(seg["start"])
            boundary_points.add(seg["end"])
        sorted_points = sorted(boundary_points)
        
        # 4. 生成无重叠子段，同时计算总负载（可选）
        for i in range(len(sorted_points) - 1):
            sub_start = sorted_points[i]
            sub_end = sorted_points[i+1]
            sub_length = sub_end - sub_start
            
            # 统计覆盖这个子段的所有原区间的负载之和
            total_load = 0
            for seg in sorted_segs:
                if seg["start"] <= sub_start and seg["end"] >= sub_end:
                    total_load += seg["load"]
            
            final_result.append({
                "segment_id": seg_id,
                "start": sub_start,
                "end": sub_end,
                "length": sub_length,
                "total_load": total_load
            })
    
    return final_result

# 执行并查看结果
output = split_overlapping_segments(data)
for item in output:
    print(item)

运行后，ID为A的子段会是[10,20]、[20,30]、[30,40]，对应的总负载分别是5、8（5+3）、3；ID为B的是[5,10]、[10,15]、[15,20]，负载分别是2、6（2+4）、4。

如果数据存在MySQL这类数据库里，可以用递归CTE来实现，适合大数据量的离线处理：

WITH RECURSIVE seg_points AS (
    -- 初始步骤：提取每个segment_id的第一个区间的start和end
    SELECT 
        segment_id,
        start AS point,
        ROW_NUMBER() OVER(PARTITION BY segment_id ORDER BY start) AS rn
    FROM segment_data
    UNION ALL
    -- 递归步骤：依次提取后续区间的start和end
    SELECT 
        s.segment_id,
        CASE WHEN r.rn % 2 = 1 THEN s.end ELSE s.start END AS point,
        r.rn + 1 AS rn
    FROM segment_data s
    JOIN seg_points r ON s.segment_id = r.segment_id
    WHERE r.rn <= (SELECT COUNT(*) * 2 FROM segment_data WHERE segment_id = s.segment_id)
),
-- 去重并排序端点
sorted_points AS (
    SELECT DISTINCT segment_id, point
    FROM seg_points
    ORDER BY segment_id, point
),
-- 生成相邻端点对，作为子段的start和end
point_pairs AS (
    SELECT 
        segment_id,
        point AS start,
        LEAD(point) OVER(PARTITION BY segment_id ORDER BY point) AS end
    FROM sorted_points
)
-- 最终生成无重叠子段，并聚合负载
SELECT 
    p.segment_id,
    p.start,
    p.end,
    p.end - p.start AS length,
    SUM(s.load) AS total_load
FROM point_pairs p
JOIN segment_data s ON p.segment_id = s.segment_id 
    AND s.start <= p.start 
    AND s.end >= p.end
WHERE p.end IS NOT NULL -- 过滤掉最后一个无匹配的端点
GROUP BY p.segment_id, p.start, p.end
ORDER BY p.segment_id, p.start;

• 区间的开闭要统一：代码默认是闭区间[start, end]，如果你的数据是左闭右开[start, end)，生成子段时要调整端点判断逻辑，比如把seg["end"] >= sub_end改成seg["end"] > sub_end
• 负载聚合可自定义：如果不需要求和，换成MAX(s.load)或AVG(s.load)就行
• 大数据量优化：Python代码可以用二分查找替代遍历，提升效率；SQL要给segment_id加索引，避免全表扫描拖慢速度

内容的提问来源于stack exchange，提问作者F.gh