Python拆分超大文件为小文件：打开文件过多问题

这种1TB级大CSV拆分的坑我踩过不少——要么频繁开关文件导致IO慢到离谱，要么一次性开太多文件触发系统句柄限制，确实两难。结合你遇到的问题，给你几个经过生产环境验证的优化方案：

方案1：文件句柄池（复用+LRU淘汰） #

核心思路是维护一个固定大小的活跃文件句柄池，不一次性打开所有目标文件，而是动态复用：当需要写入某个文件时，如果它已经在池里就直接用；不在的话，关闭池里最久没被使用的文件，再打开新的加入池。这样既避免了频繁开关文件的开销，又把同时打开的文件数控制在系统能承受的范围（比如50-100个）。

举个Python实现的例子，用OrderedDict来跟踪文件的使用顺序，实现LRU淘汰：

from collections import OrderedDict

class FileHandlePool:
    def __init__(self, max_handles=50):
        self.max_handles = max_handles
        self.handles = OrderedDict()  # 有序字典，记录使用顺序

    def get_handle(self, filename):
        if filename in self.handles:
            # 移到末尾标记为最近使用
            self.handles.move_to_end(filename)
            return self.handles[filename]
        # 超过上限，关闭最久未使用的
        if len(self.handles) >= self.max_handles:
            oldest_filename, oldest_handle = self.handles.popitem(last=False)
            oldest_handle.close()
        # 打开新文件并加入池（用1MB缓冲区减少IO）
        handle = open(filename, 'a', encoding='utf-8', buffering=1024*1024)
        self.handles[filename] = handle
        return handle

    def close_all(self):
        for handle in self.handles.values():
            handle.close()
        self.handles.clear()

# 使用示例
pool = FileHandlePool(max_handles=50)
with open('large_file.csv', 'r', encoding='utf-8', buffering=1024*1024) as src:
    header = src.readline()
    # 提前给所有目标文件写入表头（如果需要，需先获取所有目标文件名列表）
    # for fn in target_filenames:
    #     with open(fn, 'w', encoding='utf-8') as f:
    #         f.write(header)
    for line in src:
        # 替换成你的分组逻辑，这里假设按第二列分组
        group_key = line.split(',')[1].strip()
        target_filename = f'{group_key}.csv'
        handle = pool.get_handle(target_filename)
        handle.write(line)
pool.close_all()

方案2：批量分组缓存写入 #

如果系统句柄限制特别严格，或者分组键数量极多，可以用内存缓存+批量写入的方式：先把同一分组的行攒到内存里，攒到一定数量（比如1000行）再一次性写入文件，写完就关闭。这样既减少了文件打开/关闭的次数，又不会占用过多内存。

示例代码：

from collections import defaultdict

buffer_size = 1000  # 每个分组攒够1000行再写入
buffer = defaultdict(list)

with open('large_file.csv', 'r', encoding='utf-8', buffering=1024*1024) as src:
    header = src.readline()
    seen_groups = set()
    for line in src:
        # 替换成你的分组逻辑
        group_key = line.split(',')[1].strip()
        target_filename = f'{group_key}.csv'
        # 第一次遇到分组时写入表头
        if group_key not in seen_groups:
            with open(target_filename, 'w', encoding='utf-8') as f:
                f.write(header)
            seen_groups.add(group_key)
        # 加入缓存
        buffer[group_key].append(line)
        # 达到缓存阈值，批量写入
        if len(buffer[group_key]) >= buffer_size:
            with open(target_filename, 'a', encoding='utf-8') as f:
                f.writelines(buffer[group_key])
            buffer[group_key].clear()
    # 处理剩余未写入的行
    for group_key, lines in buffer.items():
        target_filename = f'{group_key}.csv'
        with open(target_filename, 'a', encoding='utf-8') as f:
            f.writelines(lines)

方案3：用系统级工具快速拆分（适合简单分组） #

如果你的分组逻辑比较简单（比如按某一列的值拆分），可以直接用Linux/macOS下的awk命令，它处理大文件的效率比Python高得多，而且内部会自动优化文件句柄的使用：

# 按第一列拆分，自动处理文件句柄，每10000行清理一次旧句柄
awk -F ',' '
    NR==1 {header=$0; next}
    {
        filename = $1 ".csv"
        if (!seen[filename]) {
            print header > filename
            seen[filename] = 1
        }
        print > filename
        # 每处理10000行关闭一次旧文件，避免句柄耗尽
        if (NR % 10000 == 0) {
            for (f in seen) close(f)
        }
    }
' large_file.csv

额外注意事项 #

• 备份原文件：操作前一定要备份，避免拆分过程中数据损坏；
• 缓冲区优化：无论是Python还是系统命令，都尽量使用较大的缓冲区（比如1MB以上），减少磁盘IO次数；
• 极端分组场景：如果分组键数量达到百万级以上，建议先对分组键做哈希取模，先拆成几十个大批次，再每个批次内细分，避免内存溢出或文件管理混乱。

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