Python基于子串去重的大文件处理代码性能优化咨询

我看了你处理1GB大文件重复行的Python代码，跑10分钟确实有不少可以提速的空间，咱们从几个核心点入手优化：

1. 立刻砍掉拖慢速度的打印操作 #

你代码里的print(":Key: " + str(key))是性能杀手！打印操作属于IO操作，比内存中的数据处理慢几个数量级，处理百万级行的时候，这个打印会把整体速度拖垮。直接删掉这行，调试完就别留着了。

2. 优化内存占用——别把所有行都存进内存 #

当前代码用defaultdict(list)把所有行都存在内存里，1GB文件如果每行平均100字节，就是1000万行，内存占用会非常高，甚至触发系统的内存交换（swap），这会让速度骤降。

改成两次遍历文件的思路：

• 第一次遍历：只统计每个key的出现次数，不存储行内容，内存占用从O(n)降到O(k)（k是不同key的数量，远小于总行数n）
• 第二次遍历：根据第一次统计的计数，直接把行写入对应的输出文件，全程只加载当前行到内存

用collections.Counter来统计计数就非常合适。

3. 优化字符串处理——少做不必要的操作 #

当前的line.split(";")会把整行分割成列表，但我们只需要第一个分号前的前12个字符，完全没必要分割整个行：

• 用line.find(";")找到第一个分号的位置，直接截取前12个字符，比split快很多
• 如果输入行首尾没有多余空白，line.strip()也可以去掉，减少不必要的字符串处理

4. 优化文件IO——减少磁盘写入次数 #

Python文件对象默认有缓冲区，但可以手动设置更大的缓冲区（比如1MB），减少磁盘IO的次数，提升写入速度。比如open(..., buffering=1024*1024)。

优化后的完整代码 #

import os
import sys
import time
from collections import Counter

def main():
    start_time = time.perf_counter()
    file_in = sys.argv[1]
    # 更清晰的输出文件名
    unique_file = f"{file_in}.proc"
    duplicate_2x_file = f"{file_in}.proc2"
    duplicate_more_file = f"{file_in}.proc3"

    # 第一次遍历：统计每个key的出现次数
    key_counter = Counter()
    with open(file_in, "r") as f:
        for line in f:
            semicolon_pos = line.find(";")
            if semicolon_pos == -1:
                # 处理没有分号的行，这里选择跳过，可根据需求调整
                continue
            # 取第一个分号前的前12个字符作为key
            key = line[:min(semicolon_pos, 12)]
            key_counter[key] += 1

    # 第二次遍历：根据计数写入对应文件
    with open(file_in, "r") as f, \
         open(unique_file, "w", buffering=1024*1024) as f1, \
         open(duplicate_2x_file, "w", buffering=1024*1024) as f2, \
         open(duplicate_more_file, "w", buffering=1024*1024) as f3:
        
        total_lines = 0
        for line in f:
            total_lines += 1
            semicolon_pos = line.find(";")
            if semicolon_pos == -1:
                continue
            key = line[:min(semicolon_pos, 12)]
            count = key_counter[key]
            if count == 1:
                f1.write(line)
            elif count == 2:
                f2.write(line)
            else:
                f3.write(line)
    
    end_time = time.perf_counter()
    print(f"处理完成，共读取 {total_lines} 行，耗时 {end_time - start_time:.2f} 秒")

if __name__ == "__main__":
    main()

额外的小优化建议 #

• 变量命名尽量清晰，比如把file_ot改成unique_file，后续维护代码更方便
• 用f-string替代字符串拼接，比如f"{file_in}.proc"比str(file_in) + ".proc"更高效也更易读
• 如果你的系统支持多线程/多进程，还可以考虑用concurrent.futures来并行处理，但对于单文件的逐行处理，两次遍历的优化已经足够让速度提升数倍（大概率能把10分钟压缩到1分钟以内）

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