如何用Python高效处理超5000万行的文本文件？

哇，5000万行的文本文件，逐行处理确实慢得让人挠头！结合你提到的已知所有60种开头两位代码这个关键信息，我整理了几个实战性拉满的Python优化方案，亲测能大幅提升处理速度：

大部分大文件处理慢的根源不是逻辑本身，而是频繁的IO调用。试试这两种方式：

• 内存映射文件（mmap）：把整个文件直接映射到内存中，相当于直接操作内存对象，避免了频繁的磁盘读写系统调用，速度提升非常明显。

import mmap

# 注意根据你的文件编码调整decode的参数
with open('your_huge_file.txt', 'r+b') as f:
    with mmap.mmap(f.fileno(), length=0, access=mmap.ACCESS_READ) as mm:
        # 按行读取内存映射的内容
        for line in iter(mm.readline, b''):
            line_str = line.decode('utf-8').strip()
            code = line_str[:2]
            # 后续处理逻辑

• 分块读取：如果内存不够映射整个文件，就每次读取一大块（比如10万行），减少系统调用次数。比逐行读for line in f:高效得多。

CHUNK_SIZE = 100000  # 可根据你的内存大小调整，越大越好（不超内存就行）

with open('your_huge_file.txt', 'r') as f:
    while True:
        # 一次性读取CHUNK_SIZE行
        chunk = f.readlines(CHUNK_SIZE)
        if not chunk:
            break
        # 批量处理这个块里的所有行
        process_chunk(chunk)

既然你已经知道所有60种代码，我们可以先把同代码的行批量归组，再统一处理，避免每行都做一次代码判断的开销：

from collections import defaultdict

# 先提前定义好每种代码对应的处理函数，比如：
def process_ab(lines):
    # 批量处理所有AB开头的行，比如解析、计算、写入等
    for line in lines:
        # 具体逻辑...

def process_bc(lines):
    # BC代码的处理逻辑...

# 把函数和代码映射成字典，方便快速查找
code_handlers = {
    'AB': process_ab,
    'BC': process_bc,
    # ... 剩下的58种代码都对应好
}

def process_chunk(chunk):
    # 先把块里的行按代码分组
    grouped_lines = defaultdict(list)
    for line in chunk:
        code = line[:2]
        grouped_lines[code].append(line)
    
    # 批量处理每个分组
    for code, lines in grouped_lines.items():
        code_handlers[code](lines)

这种方式在同代码行数量多的时候，能省掉大量重复的条件判断时间。

如果你的处理逻辑是CPU密集型（比如大量计算、解析），用多进程绕开GIL限制；如果是IO密集型（比如处理后要写多个文件），用多线程或者异步IO就行。

这里用concurrent.futures.ProcessPoolExecutor举个例子：

from concurrent.futures import ProcessPoolExecutor

# 注意：处理函数要能被序列化（pickle），所以尽量不要用嵌套函数
def process_group(args):
    code, lines = args
    code_handlers[code](lines)

if __name__ == '__main__':
    CHUNK_SIZE = 100000
    MAX_WORKERS = 4  # 设为你的CPU核心数，比如8核就设8

    with open('your_huge_file.txt', 'r') as f:
        with ProcessPoolExecutor(max_workers=MAX_WORKERS) as executor:
            while True:
                chunk = f.readlines(CHUNK_SIZE)
                if not chunk:
                    break
                # 先分组
                grouped = defaultdict(list)
                for line in chunk:
                    code = line[:2]
                    grouped[code].append(line)
                # 把每个分组的任务提交给进程池
                executor.map(process_group, grouped.items())

⚠️ 注意：如果需要写入文件，要么让每个进程写独立的输出文件（最后再合并），要么用multiprocessing.Lock加锁，避免写入冲突。

如果你的每行是空格/分隔符分隔的结构化数据，Pandas的底层是C实现的，批量处理速度比纯Python循环快N倍：

import pandas as pd

# 分块读取文件，chunksize根据内存调整
chunk_iter = pd.read_csv(
    'your_huge_file.txt',
    sep=' ',  # 按空格分隔，根据你的实际分隔符调整
    header=None,  # 没有表头
    chunksize=100000
)

for chunk in chunk_iter:
    # 按第一列（也就是开头的两位代码）分组
    grouped = chunk.groupby(chunk[0])
    for code, group in grouped:
        # 把DataFrame传给处理函数，比如提取列、计算等
        code_handlers[code](group)

如果你的处理逻辑能适配Pandas的向量操作，速度会更快。

• 预编译正则：如果需要用正则解析行内容，一定要提前编译，不要每次处理行时再编译：

import re
# 预编译正则表达式
line_pattern = re.compile(r'^([A-Z]{2}) (\d+) (\d+) (-?\d+)')

def process_ab(lines):
    for line in lines:
        match = line_pattern.match(line)
        if match:
            code, num1, num2, num3 = match.groups()
            # 处理逻辑...

• 避免字符串复制：处理行时尽量用切片（比如line[:2]）而非split()，减少不必要的内存分配。

最后提醒：先拿小样本测试每个方案的效果，调整CHUNK_SIZE、MAX_WORKERS这些参数，找到最适合你机器的配置。

内容的提问来源于stack exchange，提问作者KUNAL SHARMA