用 heapq 在流式日志中维护耗时 Top K 接口

问题场景

服务会产生大量访问日志，每条日志通常包含接口路径和耗时。若先把全部请求读入列表再排序，内存会随日志量线性增长；在容器内存受限或日志持续写入的场景下并不合适。目标是从流式日志中只保留耗时最高的 K 条记录，内存占用控制在 O(K)。

核心思路

维护一个固定大小为 K 的最小堆：

• 堆未满时，新记录直接入堆；
• 堆满后，只有新记录耗时严格大于堆顶（当前 Top K 中耗时最小的记录）才替换堆顶；
• 日志处理结束后，对堆内元素排序输出。

这样堆内始终保存当前已见记录中的 Top K 候选。使用 Python 标准库 heapq 即可，时间复杂度为 O(n log K)，空间复杂度为 O(K)。这里不保存全量请求，也不对全量数据排序。

并列值处理

耗时相同的记录需要明确取舍。下面用递增序号 seq 作为次级键。最小堆的堆顶是“最差”的记录：耗时小者优先；耗时相同时，序号更大（更晚出现）者优先被淘汰。替换条件使用严格大于，因此并列耗时不会触发替换，最终保留更早出现的记录。输出时再按耗时降序、序号升序排序，保证结果稳定可读。

import heapq
from dataclasses import dataclass

@dataclass
class Record:
    cost_ms: float
    path: str
    seq: int

    def __lt__(self, other: 'Record') -> bool:
        if self.cost_ms != other.cost_ms:
            return self.cost_ms < other.cost_ms
        return self.seq > other.seq

class TopKSlow:
    def __init__(self, k: int):
        if k <= 0:
            raise ValueError('k 必须为正整数')
        self.k = k
        self._heap = []
        self._seq = 0

    def add(self, path: str, cost_ms: float) -> None:
        rec = Record(cost_ms, path, self._seq)
        self._seq += 1
        if len(self._heap) < self.k:
            heapq.heappush(self._heap, rec)
        elif rec.cost_ms > self._heap[0].cost_ms:
            heapq.heapreplace(self._heap, rec)

    def result(self):
        return sorted(self._heap, key=lambda r: (-r.cost_ms, r.seq))

读取日志行

下面把每行 JSON 解析成路径和耗时。空行、格式错误、字段缺失或类型不匹配的行直接跳过，避免单条坏日志中断整个任务。示例字段使用占位值，实际字段名可按日志格式调整。

import json
from typing import Iterable, Iterator, Tuple

def parse_lines(lines: Iterable[str]) -> Iterator[Tuple[str, float]]:
    for line in lines:
        line = line.strip()
        if not line:
            continue
        try:
            obj = json.loads(line)
            path = str(obj['path'])
            cost_ms = float(obj['cost_ms'])
        except (json.JSONDecodeError, KeyError, TypeError, ValueError):
            continue
        yield path, cost_ms

def top_k_from_log(lines: Iterable[str], k: int = 3):
    tracker = TopKSlow(k)
    for path, cost_ms in parse_lines(lines):
        tracker.add(path, cost_ms)
    return tracker.result()

运行方法

保存为 topk_log.py，使用 Python 3 运行。示例用 json.dumps 构造占位日志行，避免手写 JSON 时出现转义错误：

if __name__ == '__main__':
    sample = [
        {'path': '/api/a', 'cost_ms': 120},
        {'path': '/api/b', 'cost_ms': 350},
        {'path': '/api/c', 'cost_ms': 80},
        {'path': '/api/d', 'cost_ms': 350},
        {'path': '/api/e', 'cost_ms': 500},
        {'path': '/api/f', 'cost_ms': 20},
        {'path': '/api/g', 'cost_ms': 500},
    ]
    lines = [json.dumps(item) for item in sample]
    for i, rec in enumerate(top_k_from_log(lines, k=3), 1):
        print(i, rec.path, rec.cost_ms)

输出：

1 /api/e 500.0
2 /api/g 500.0
3 /api/b 350.0

两个 350 毫秒的记录中，/api/d 更晚出现，在并列规则下优先级更低，会先位于堆顶并可能被后续更高耗时记录替换；因此最终保留更早出现的 /api/b。这体现了严格大于才替换的取舍。

边界与限制

• k 必须为正整数；k 大于实际记录数时，输出所有已见记录；输入为空时返回空列表。
• 堆内只保留 K 条候选，result() 返回前必须排序，因为 heapq 不维护整体有序。
• 浮点耗时存在精度问题，生产环境可改为整数微秒或整数毫秒，避免相等判断受浮点误差影响。
• 当前实现按请求计数，同一路径多次出现会在 Top K 中重复。若需要按接口聚合，应先用字典维护每个接口的最大耗时或分位耗时，再对聚合结果维护堆。
• 跳过坏行会掩盖数据质量问题，建议增加计数并上报，而不是静默丢弃。
• 本方案适用于单进程内有界内存的流式统计；多实例日志需要各实例分别计算后再合并，或交给集中式流处理组件。