如何加速大文件按城市分文件写入的处理流程？

嘿，看了你的代码和问题描述，核心瓶颈确实在频繁打开/关闭文件上——磁盘读写本来就是程序里最慢的环节之一，你每次处理一行就打开、写入、关闭对应城市的文件，相当于把磁盘IO次数放大到了行数级别，这也是为什么80%的时间都耗在这了。下面给你几个实打实的优化方案，从最关键的改进到细节调优都有，应该能帮你把处理速度提上去不少：

1. 最立竿见影的优化：复用文件句柄 #

不要每次处理一行就开关文件，而是维护一个字典保存每个城市对应的文件对象，全程只打开一次，处理完所有行后统一关闭。这样能把磁盘IO次数从几万/几十万次直接降到城市数量级，性能提升非常明显。

优化后的核心代码示例：

import re
import os
from collections import defaultdict

def processFile(file):
    # 正则预编译已经做了，这里换成match更高效（因为是整行匹配）
    pattern = re.compile(r"(\d{4}-\d{2}-\d{2})\t(\d{2}:\d{2})\t(.+)\t(.+)\t(\d+\.\d+|\d+)\t(\w+)\n")
    total_sale = 0
    # 用defaultdict简化字典更新操作
    city_dict = defaultdict(float)
    categories_dict = defaultdict(float)
    os.makedirs("report", exist_ok=True)
    
    # 维护「城市名 -> 文件句柄」的映射
    city_file_handles = {}
    
    try:
        # 用with自动管理输入文件，避免手动close出错
        with open(file, 'r') as input_f:
            for line in input_f:
                valid_entry = pattern.match(line)
                if not valid_entry:
                    print(f"Invalid entry: '{line.strip()}'")
                    continue
                
                entry_sale = float(valid_entry.group(5))
                total_sale += entry_sale
                city = valid_entry.group(3)
                category = valid_entry.group(4)
                
                # 更新统计字典（比update更高效）
                city_dict[city] += entry_sale
                categories_dict[category] += entry_sale
                
                # 写入对应城市文件：复用已打开的句柄
                if city not in city_file_handles:
                    # 第一次遇到该城市时才打开文件，用追加模式不用判断文件是否存在
                    file_path = f"report/{city}.txt"
                    city_file_handles[city] = open(file_path, 'a')
                # 直接写原行，不用取group(0)（group(0)就是原行内容）
                city_file_handles[city].write(line)
    finally:
        # 确保所有文件句柄都关闭，避免资源泄漏
        for handle in city_file_handles.values():
            handle.close()
    
    return (dict(city_dict), dict(categories_dict), total_sale)

2. 进一步提升：批量写入缓存 #

如果你的城市数量极多，或者单个城市的行数特别多，可以再加上批量缓存：先把每行内容暂存到对应城市的列表中，等缓存到一定大小（比如1000行）再一次性写入，进一步减少磁盘IO的触发次数。

核心代码片段（替换上面的写入逻辑）：

# 用字典维护每个城市的内容缓存
city_buffers = defaultdict(list)
# 设定缓存阈值，根据你的内存情况调整
BUFFER_THRESHOLD = 1000

# 循环内的写入逻辑改为：
city_buffers[city].append(line)
# 当缓存达到阈值时批量写入
if len(city_buffers[city]) >= BUFFER_THRESHOLD:
    if city not in city_file_handles:
        file_path = f"report/{city}.txt"
        city_file_handles[city] = open(file_path, 'a')
    # 用join一次性拼接再写入，比循环写更快
    city_file_handles[city].write(''.join(city_buffers[city]))
    city_buffers[city].clear()

# 循环结束后，写入剩余的缓存内容
for city, lines in city_buffers.items():
    if lines:
        if city not in city_file_handles:
            file_path = f"report/{city}.txt"
            city_file_handles[city] = open(file_path, 'a')
        city_file_handles[city].write(''.join(lines))

3. 其他细节优化 #

• 用match代替search：因为你的正则是匹配整行内容，match会从字符串开头开始匹配，比search扫描整个字符串更高效。
• 使用collections.defaultdict：避免每次调用dict.get()，代码更简洁且性能略优。
• 直接写入原行：原代码中valid_entry.group(0)就是原行内容，直接用line代替即可，省去正则分组的额外开销。
• 避免不必要的判断：用文件追加模式'a'可以直接覆盖“文件是否存在”的判断，不存在会自动创建，存在则追加内容。

这些优化组合起来，应该能把文件写入的时间占比降到非常低，整体处理速度会有质的提升。

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