Python 日志词频统计实战：用 collections.Counter 一行代码找出高频错误

背景

排查线上问题时，日志文件动辄几十万行，靠肉眼翻根本翻不完。其实多数时候我们只关心一个问题：哪种错误出现得最频繁？把“数一数每种错误出现了几次”交给程序，几秒钟就能定位主要矛盾。Python 标准库的 collections.Counter 就是为“计数”而生的容器：传一个可迭代对象进去，它自动统计每个元素出现的次数，还自带取 Top N、合并、相减等方法，比手写 dict 累加简洁得多。

实现思路

分三步走。第一步，逐行读日志，只保留级别为 ERROR 的行，避免 INFO、DEBUG 的噪音干扰统计。第二步，用简单的规则从每行提取错误关键词——这里演示用正则取方括号里的错误码（如 [TIMEOUT]、[DB_CONN_FAIL]），真实项目里也可以按异常类名或接口名提取。第三步，把关键词列表交给 Counter，调用 most_common(5) 直接拿到出现次数最多的前 5 种错误；total() 可以看错误总数。如果日志分布在多台机器上，各自统计后用 Counter 相加就能汇总，Counter 支持直接用 + 合并。

需要注意两点：一是 most_common() 返回的是按次数从高到低排好的 (元素, 次数) 元组列表，开箱即用；二是 Counter 对不存在的键返回 0 而不是抛 KeyError，所以 counter[“某个没出现过的错误”] 得到 0，可以放心做减法或比较，不用先判断键在不在。

代码

import re
from collections import Counter

SAMPLE_LOGS = [
    "2026-11-06 09:00:01 ERROR [TIMEOUT] 调用支付接口超时 3000ms",
    "2026-11-06 09:00:02 INFO 用户登录成功 user-7",
    "2026-11-06 09:00:03 ERROR [DB_CONN_FAIL] 数据库连接池耗尽",
    "2026-11-06 09:00:04 ERROR [TIMEOUT] 调用支付接口超时 3000ms",
    "2026-11-06 09:00:05 ERROR [TIMEOUT] 调用短信接口超时 3000ms",
    "2026-11-06 09:00:06 WARN 磁盘使用率 82%",
    "2026-11-06 09:00:07 ERROR [NULL_POINTER] 订单服务空指针",
    "2026-11-06 09:00:08 ERROR [TIMEOUT] 调用支付接口超时 3000ms",
    "2026-11-06 09:00:09 ERROR [DB_CONN_FAIL] 数据库连接池耗尽",
    "2026-11-06 09:00:10 ERROR [TIMEOUT] 调用风控接口超时 3000ms",
]

ERROR_CODE = re.compile(r"\[([A-Z_]+)\]")


def count_errors(log_lines):
    """统计 ERROR 行里的错误码出现次数，返回 Counter。"""
    codes = []
    for line in log_lines:
        if "ERROR" not in line:
            continue
        match = ERROR_CODE.search(line)
        if match:
            codes.append(match.group(1))
    return Counter(codes)


def main() -> None:
    counter = count_errors(SAMPLE_LOGS)
    print(f"错误总数: {counter.total()}")
    print("出现次数最多的错误 Top 3:")
    for code, times in counter.most_common(3):
        print(f"  {code}: {times} 次")
    print(f"未出现过的错误码计数: {counter['SOME_NEVER_SEEN']}")
    # 多台机器的统计结果可以直接相加汇总
    other_machine = Counter({"TIMEOUT": 2, "DB_CONN_FAIL": 1})
    merged = counter + other_machine
    print(f"汇总后 TIMEOUT 总数: {merged['TIMEOUT']}")


if __name__ == "__main__":
    main()

验证结果

已在本地 Python 3.12 实测。10 行样本日志里共 8 行 ERROR，错误总数为 8；Top 3 依次是 TIMEOUT 出现 5 次、DB_CONN_FAIL 出现 2 次、NULL_POINTER 出现 1 次，与样本内容逐行核对一致。未出现过的错误码计数为 0，未抛异常；两台机器的 Counter 相加后，TIMEOUT 汇总为 7（5 + 2），合并结果正确。

注意事项

• 先过滤再统计：INFO、DEBUG 行不参与计数，否则高频的正常日志会淹没真正的错误信号。
• 提取规则要稳定：错误码、异常类名这类结构化字段适合做统计键；直接对整行文本计数会因为时间戳、参数不同而把同一类错误拆成几百条。
• Counter 相加会丢弃计数为 0 或负数的项：做减法对比“修复前后”时要注意，消失的错误码不会出现在结果里，这是符合预期的。
• 数据量大时不要一次性读入整个文件：逐行迭代、边读边喂给 Counter.update()，内存占用只和错误种类数有关，和日志总行数无关。
• 统计只是定位问题的第一步：拿到 Top 1 错误后，还要结合具体日志上下文确认根因，不要只看数字就下结论。