Python处理大日志文件的三种方案及对比详解_Python

背景与痛点

运维排查线上问题时，经常需要从 gb 级别的应用日志中快速提取关键字（如 error、特定 trace_id）。传统做法是用 grep 或 python 逐行读取，但当日志达到几个 gb 时，python 的 readline 循环会变得异常缓慢，内存占用也居高不下。本文通过一个真实案例，对比三种实现方案，并给出性能验证方法。

方案一：逐行流式读取（baseline）

最直观的实现是使用 open() 配合 for line in f，逐行扫描。代码如下：

def filter_stream(log_path, keyword, out_path):
    with open(log_path, 'r', encoding='utf-8', errors='ignore') as fin, \
         open(out_path, 'w') as fout:
        for line in fin:
            if keyword in line:
                fout.write(line)

思路：简单可靠，但每次迭代都涉及 python 层的字符串解码和关键字匹配，对于 2gb 日志，实测耗时约 45 秒，内存占用约 20mb（主要缓冲）。

方案二：mmap + 内存视图（优化）

优化思路：利用 mmap 将文件映射到虚拟内存，避免频繁 read 系统调用；再通过 bytes.find() 定位关键字，并配合 memoryview 切片输出。核心代码：

import mmap
def filter_mmap(log_path, keyword, out_path):
    keyword_b = keyword.encode()
    with open(log_path, 'rb') as f, \
         mmap.mmap(f.fileno(), 0, access=mmap.access_read) as mm, \
         open(out_path, 'wb') as fout:
        start = 0
        while true:
            idx = mm.find(keyword_b, start)
            if idx == -1:
                break
            # 找到行首和行尾
            line_start = mm.rfind(b'\n', 0, idx) + 1
            line_end = mm.find(b'\n', idx)
            if line_end == -1:
                line_end = mm.size()
            fout.write(mm[line_start:line_end])
            fout.write(b'\n')
            start = line_end + 1

要点：mmap.find 是 c 实现的，速度远超 python 层循环；但需注意处理边界情况（关键字跨行、最后一行无换行符）。

方案三：多进程分块并行（终极优化）

当单机多核时，可将文件按字节偏移均分成 n 块，每个进程独立处理一块，最后合并结果。为避免切块时截断行，需要查找块首的换行符对齐。这里给出简化示例：

import os, multiprocessing as mp
def worker(chunk_path, start, end, keyword, out_path):
    with open(chunk_path, 'rb') as f:
        f.seek(start)
        # 对齐到行首（略）
        data = f.read(end - start)
        lines = data.split(b'\n')
        matched = [line for line in lines if keyword in line]
        with open(out_path, 'ab') as fout:
            for line in matched:
                fout.write(line + b'\n')
def parallel_filter(log_path, keyword, num_workers=4):
    size = os.path.getsize(log_path)
    chunk_size = size // num_workers
    # 创建任务列表（略）

实际应用时需处理块边界，可用 seek 定位到下一个换行符。并行方案在 8 核机器上可将 2gb 日志处理时间从 45 秒降至 8 秒。