Python 生成器处理大文件实战：逐行读取、过滤和分批写入

Python 处理文本文件很方便，但如果遇到几百 MB 甚至几个 GB 的日志、CSV 或导出数据，一句 read() 或 readlines() 就可能让内存一路飙升。脚本在本地小样本上没问题，换到生产数据就开始变慢、卡住，甚至被系统杀掉。

处理大文件的核心思路是：让数据一行一行流过程序，不要一次性搬进内存。本文用日志清洗场景，把逐行读取、过滤有效行、按批次写入结果串成一个稳定的数据管道。

摘要

本文会先复现一次性读取大文件的内存问题，再使用 Python 生成器实现流式处理：读取器逐行产出文本，过滤器只保留目标行，批处理函数按固定数量写入结果。最后补充编码、异常行、批次大小和进度日志的注意点。

适合人群

• 正在用 Python 清洗日志、CSV、导出文件的开发者。
• 遇到脚本内存占用过高，希望改成流式处理的同学。
• 已经掌握基础文件读写，想理解生成器落地场景的读者。

目录

1. 一次性读取为什么危险
2. 用生成器逐行读取
3. 在管道中做过滤和转换
4. 按批次写入结果
5. 常见坑和优化建议
6. 总结

一、一次性读取为什么危险

很多脚本最初都会这样写：

from pathlib import Path

path = Path("access.log")

lines = path.read_text(encoding="utf-8").splitlines()
error_lines = [line for line in lines if "ERROR" in line]

Path("errors.log").write_text("\n".join(error_lines), encoding="utf-8")

这段代码对小文件很顺手，但它至少会在内存里放三份东西：完整文本、拆分后的行列表、过滤后的结果列表。文件越大，内存压力越明显。更糟的是，如果某一行异常、编码不统一，整个脚本可能在中间失败，已经处理过的进度也不好保留。

二、用生成器逐行读取

更稳的方式是用生成器逐行产出内容。调用方不关心文件怎么读，只关心“下一行是什么”。

from pathlib import Path
from typing import Iterator

def iter_lines(path: Path) -> Iterator[str]:
    with path.open("r", encoding="utf-8", errors="replace") as file:
        for line in file:
            yield line.rstrip("\n")

for line in iter_lines(Path("access.log")):
    if "ERROR" in line:
        print(line)

yield 的好处是懒加载：程序每次只拿一行，处理完再拿下一行。即使文件很大，内存里常驻的也只是当前行和少量中间状态。

三、在管道中做过滤和转换

生成器很适合把处理流程拆成一段一段。比如我们只保留包含 ERROR 的日志，并把前后空格去掉：

from typing import Iterable, Iterator

def keep_errors(lines: Iterable[str]) -> Iterator[str]:
    for line in lines:
        text = line.strip()
        if "ERROR" in text:
            yield text

def add_prefix(lines: Iterable[str]) -> Iterator[str]:
    for index, line in enumerate(lines, start=1):
        yield f"{index}: {line}"

这样写的优势是每个函数只做一件事。后面如果要改成过滤 WARN、提取 IP、清理空行，都能局部调整。

四、按批次写入结果

如果每处理一行就写一次文件，磁盘写入次数会太多；如果全部攒到列表里再写，又回到了内存问题。折中方式是按批次写入。

from pathlib import Path
from typing import Iterable, Iterator, List

def chunked(lines: Iterable[str], size: int) -> Iterator[List[str]]:
    batch: List[str] = []
    for line in lines:
        batch.append(line)
        if len(batch) >= size:
            yield batch
            batch = []

    if batch:
        yield batch

def write_batches(lines: Iterable[str], output: Path, size: int = 1000) -> int:
    count = 0
    with output.open("w", encoding="utf-8") as file:
        for batch in chunked(lines, size):
            file.write("\n".join(batch))
            file.write("\n")
            count += len(batch)
    return count

把前面的函数串起来，就得到一个完整脚本：

from pathlib import Path

source = Path("access.log")
target = Path("errors.log")

lines = iter_lines(source)
errors = keep_errors(lines)
numbered = add_prefix(errors)

total = write_batches(numbered, target, size=1000)
print(f"done, wrote {total} lines")

这条管道里，数据从源文件逐行流向目标文件，中间没有巨大的列表。批次大小可以根据磁盘性能和内存情况调整。

五、常见坑和优化建议

1. 编码问题要提前处理

日志文件可能混入非法字符。读取时使用 errors="replace" 可以让脚本继续跑，并把异常字符替换掉。对强一致场景，也可以记录异常行后再人工处理。

2. 不要在管道中间转成 list

一旦写出 list(iter_lines(path))，就会把所有行重新装进内存。生成器管道的价值在于“用一行、处理一行、释放一行”。

3. 批次大小要按场景调

批次太小，写文件次数多；批次太大，内存占用会上升。文本日志一般可以从 500 到 5000 行试起，再根据文件大小和机器配置调整。

4. 给长任务加进度日志

大文件处理可能跑几分钟甚至更久。可以每处理 10000 行打印一次进度，方便判断脚本是否还在正常推进。

def report_progress(lines: Iterable[str], step: int = 10000) -> Iterator[str]:
    for index, line in enumerate(lines, start=1):
        if index % step == 0:
            print(f"processed {index} lines")
        yield line

六、总结

Python 处理大文件时，最重要的不是写出复杂代码，而是避免一次性把所有内容读进内存。用生成器把读取、过滤、转换、批量写入拆成独立步骤，数据会像流水一样经过每个环节。

落地时记住三点：文件逐行读，中间不转列表，结果按批次写。这样即使数据量从几 MB 增长到几个 GB，脚本也能保持稳定、可观察、容易扩展。