如何按小时对跨时段运行的进程记录进行分组统计

这个需求其实是典型的时间区间重叠聚合统计问题，核心就是要判断每个进程的运行窗口和我们要统计的小时窗口是否有交集，然后对每个小时统计交集的进程数量。我来给你拆解下思路和具体实现方案：

核心判断逻辑 #

对于任意一个统计小时（比如2019/01/01 14:00-15:00），一个进程需要被计数的条件是：

• 进程的启动时间 早于 该小时的结束时间（15:00）
• 进程的结束时间 晚于 该小时的开始时间（14:00）

用这个逻辑套你的示例数据：

• 14点小时窗口：A（14:10-14:55）、B（14:20-16:30）都满足，计数2
• 15点小时窗口：B（14:20-16:30）、C（15:05-15:10）都满足，计数2
• 16点小时窗口：只有B（14:20-16:30）满足（结束时间16:30晚于16:00，启动时间早于17:00），计数1
  完全符合你期望的结果。

方案1：SQL实现（适合数据库存储的日志） #

如果你的进程日志存在关系型数据库里，用CTE生成小时序列再关联统计是最高效的方式。假设表名为process_logs，字段为process_name、start_time（datetime类型）、end_time（datetime类型）：

-- 生成需要统计的所有小时区间（从最早进程启动的整点到最晚进程结束的整点）
WITH hours AS (
    SELECT 
        generate_series(
            date_trunc('hour', MIN(start_time)),
            date_trunc('hour', MAX(end_time)),
            '1 hour'::interval
        ) AS hour_start
    FROM process_logs
)
SELECT
    EXTRACT(HOUR FROM h.hour_start) AS hour,
    COUNT(DISTINCT p.process_name) AS process_count
FROM hours h
LEFT JOIN process_logs p 
    ON p.start_time < h.hour_start + INTERVAL '1 hour'
    AND p.end_time > h.hour_start
GROUP BY h.hour_start
ORDER BY h.hour_start;

代码说明：

• generate_series函数（PostgreSQL支持，MySQL可以用递归CTE实现类似逻辑）生成连续的小时区间
• 关联条件就是我们前面说的重叠判断逻辑
• COUNT(DISTINCT)用来避免同名进程重复计数，如果你的进程有唯一ID，用ID代替process_name会更准确

方案2：Python实现（适合本地文件/内存数据） #

如果你的数据是CSV或者存在内存里，用Python原生datetime处理就可以，这里给你实现的版本：

from datetime import datetime, timedelta

# 示例进程数据
process_list = [
    {"name": "A", "start": "2019/01/01 14:10", "end": "2019/01/01 14:55"},
    {"name": "B", "start": "2019/01/01 14:20", "end": "2019/01/01 16:30"},
    {"name": "C", "start": "2019/01/01 15:05", "end": "2019/01/01 15:10"},
]

# 把字符串时间转成datetime对象
for proc in process_list:
    proc["start"] = datetime.strptime(proc["start"], "%Y/%m/%d %H:%M")
    proc["end"] = datetime.strptime(proc["end"], "%Y/%m/%d %H:%M")

# 生成所有需要统计的小时区间
min_start = min(p["start"] for p in process_list)
max_end = max(p["end"] for p in process_list)
# 取最早的整点时间
current_hour = min_start.replace(minute=0, second=0, microsecond=0)
hour_windows = []
while current_hour <= max_end:
    hour_windows.append(current_hour)
    current_hour += timedelta(hours=1)

# 统计每个小时的进程数
result = []
for hour_start in hour_windows:
    hour_end = hour_start + timedelta(hours=1)
    count = 0
    for proc in process_list:
        # 判断进程和当前小时是否重叠
        if proc["start"] < hour_end and proc["end"] > hour_start:
            count += 1
    result.append({
        "hour": hour_start.hour,
        "process_count": count
    })

# 输出结果
print("Hour ProcessCount")
for item in result:
    print(f"{item['hour']:4d} {item['process_count']:12d}")

运行这段代码就能得到你期望的输出：

Hour ProcessCount
  14            2
  15            2
  16            1

注意事项 #

• 如果进程运行跨天（比如从23:30到次日01:30），上述逻辑会自动统计23点、0点、1点三个小时的计数
• 如果有重复的进程名（比如同一个进程名重启多次），记得用进程唯一ID来做去重统计，避免计数错误
• 对于超大规模的日志数据，SQL方案的性能会远优于Python遍历，优先用数据库层面的统计

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