Python案例怎么逐行读取大文件？

wen python案例 2026-06-08 57

Python大文件逐行读取：高效处理海量数据的核心技巧

目录导读

为什么需要逐行读取大文件？
基础方法：for line in file 的底层原理
进阶技巧：按块读取与迭代器优化
实战案例：处理1GB日志文件的性能对比
常见问题问答（FAQ）
总结与最佳实践

为什么需要逐行读取大文件？

在数据分析、日志处理或ETL任务中，经常遇到远超内存容量的文件（例如10GB+的CSV或日志文件），如果使用file.read()一次性加载，轻则内存爆炸，重则程序崩溃。逐行读取通过流式处理，每次只加载一行到内存，是处理大文件的黄金法则。

Python案例怎么逐行读取大文件？

案例：某电商平台每天生成5GB的访问日志，需要提取所有“404”状态码的行，若用readlines()读取，内存会瞬间耗尽；而逐行读取只需占用几KB内存。

基础方法：`for line in file` 的底层原理

Python最优雅的逐行读取方式就是：

with open('huge_file.log', 'r', encoding='utf-8') as f:
    for line in f:
        process(line)  # 处理每一行

原理：for line in f实际上调用了文件对象的__next__()方法，内部使用缓冲区（默认8KB）预读数据，然后按\n分割成行，这意味着：

内存消耗 ≈ 缓冲区大小 + 单行最大长度
自动处理编码和换行符差异（Windows的\r\n会被自动转换为\n）

性能对比：

方法	内存占用（1GB文件）	速度
`readlines()`	约1GB+	快（但不可用）
`for line in f`	约几KB	中等
`read().splitlines()`	约1GB+	极快（但不可用）

注意：当一行数据特别长（例如包含嵌入式换行符的CSV字段）时，这种方法仍然会加载超长行到内存，此时需要改用read(size)配合自定义分隔符。

进阶技巧：按块读取与迭代器优化

1 使用`read(size)`按块读取

当需要更精细控制缓冲区时,可以自定义块大小：

def chunked_read(file_path, chunk_size=8192):
    with open(file_path, 'r') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk
# 使用
for chunk in chunked_read('big.csv'):
    # 处理chunk，需要自行拆分行
    lines = chunk.splitlines(keepends=True)
    last_line = ''
    for line in lines:
        if line.endswith('\n'):
            process(last_line + line)
            last_line = ''
        else:
            last_line = line  # 不完整行，等待下一个chunk

适用场景：处理二进制文件或自定义分隔符数据（如记录以特殊字符分隔）。

2 使用`linecache`实现随机访问

如果大文件需要频繁随机访问某几行（例如调试日志），不要用逐行读取：

import linecache
line = linecache.getline('huge.log', 1000000)  # 获取第100万行

原理：它会将文件索引缓存到内存中，但首次遍历仍需逐行读取。

3 并行处理：多进程逐行读取

对于CPU密集型处理,可以用multiprocessing配合file.seek实现并行：

import multiprocessing as mp
def process_chunk(file_path, start, end):
    with open(file_path, 'r') as f:
        f.seek(start)
        for line in f:
            if f.tell() >= end:
                break
            process(line)
# 主程序先获取文件总大小
file_size = os.path.getsize('huge.log')
num_workers = 4
chunk_size = file_size // num_workers
jobs = []
for i in range(num_workers):
    start = i * chunk_size
    end = (i + 1) * chunk_size if i < num_workers - 1 else file_size
    p = mp.Process(target=process_chunk, args=(file_path, start, end))
    jobs.append(p)
    p.start()

注意：必须确保每个进程的start位置是某行的开头（否则会读取不完整的一行）。

实战案例：处理1GB日志文件的性能对比

假设有一个access.log文件，大小1.2GB，需要统计“404”状态码的出现次数。

方法1：朴素逐行读取（推荐）

count = 0
with open('access.log', 'r') as f:
    for line in f:
        if ' 404 ' in line:
            count += 1
print(count)

内存占用：~8KB
耗时：约12秒（SSD）
最优解：简单、稳定、内存友好

方法2：使用`mmap`内存映射

import mmap
with open('access.log', 'r') as f:
    with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as m:
        count = 0
        for line in m:
            if b' 404 ' in line:
                count += 1
print(count)

内存占用：文件大小（1.2GB虚拟内存，但实际物理缓存由OS控制）
耗时：约8秒
适用场景：需要频繁随机读取大文件时，但对单次顺序扫描优势不明显。

方法3：`pandas`分块读取

import pandas as pd
chunks = pd.read_csv('access.log', chunksize=100000, sep=' ', header=None)
count = sum(1 for chunk in chunks if ' 404 ' in chunk[3].values)

内存占用：根据chunksize调整（约10MB）
耗时：约25秒（含类型推断开销）
如果数据需要结构化分析,牺牲性能换便利性；纯文本匹配不建议用pandas。

常见问题问答（FAQ）

Q1：逐行读取时遇到特别长的一行（比如10万字符），内存会爆吗？

A：会，一行数据的长度不能超过内存大小，解决方法：使用read(size)按固定字节块读取，并在块边界手动拼接行。

Q2：Windows系统下，\r\n是否会影响读取？

A：不会，Python的open()默认使用通用换行模式（mode='r'），会自动将所有换行转换为\n，但如果你需要保留原始换行符，请使用open(file, 'rb')二进制模式。

Q3：逐行读取100GB的文件需要多久？

A：取决于硬盘速度，SATA SSD约500MB/s，NVMe约3GB/s，机械硬盘约150MB/s，100GB文件在NVMe上约需35秒，CPU处理速度通常远快于IO，瓶颈在磁盘。

Q4：for line in f 和 while line = f.readline() 哪个更好？

A：for line in f更优，因为它是用迭代器实现，在Python底层用C语言优化，而readline()是Python函数调用，且for循环会在文件结束自动停止，无需判断空行。

Q5：如何处理文件编码问题？

A：对于utf-8文件，直接open(file, encoding='utf-8')，但有些文件混合编码，可用errors='replace'或chickadee库自动检测，建议先用chardet检测编码。

总结与最佳实践

场景	推荐方法	原因
处理文本日志	`for line in file`	最简洁，内存占用低
处理二进制/记录文件	`file.read(chunk_size)`	可控制边界分割
需要随机访问	`linecache` 或 `mmap`	避免重新扫描
CPU密集型处理	多进程+`seek`分块	线性加速
结构化数据分析	`pandas.read_csv(`chunksize)	兼顾效率和易用性