Python高效批量写文件方案验证及大文件批量读取疑问

一、核心思路的合理性 #

你的方向完全正确：用生成器避免全量数据加载到内存，用进程池提升并行写入效率，itertools.islice切分生成器的思路也没问题，但原代码存在一个致命错误需要修正。

二、原代码的关键问题 #

原循环中每次调用generate_lines()都会重新生成完整的序列，导致每个进程拿到的切片都是从头开始的重复数据，最终所有输出文件内容会完全一致，而非分段的完整数据集。

修正后的代码 #

import multiprocessing
import itertools

def generate_lines():
    for i in range(10000):
        yield f"Line {i + 1}"

def write_lines(filename, lines):
    with open(filename, 'w') as file:
        for line in lines:
            file.write(line + '\n')

if __name__ == '__main__':
    with multiprocessing.Pool(2) as pool:
        chunk_size = 5000
        # 仅初始化一次生成器，避免重复生成数据
        line_generator = generate_lines()
        chunk_index = 1
        
        while True:
            # 从同一个生成器中切分下一批数据
            chunk = list(itertools.islice(line_generator, chunk_size))
            if not chunk:
                break
            pool.apply_async(write_lines, (f'file{chunk_index}.txt', chunk))
            chunk_index += 1

        pool.close()
        pool.join()

    print("Writing completed successfully.")

这里将itertools.islice作用在同一个生成器实例上，每次切片会消耗生成器的对应部分，不会重复生成数据；同时将切片转为列表，解决了生成器无法跨进程序列化传递的问题，且chunk大小可控，内存占用依然保持在低水平。

三、大源文件的生成器批量读取实现 #

完全可以用生成器实现大文件的批量读取，既避免全量加载，又能按批次处理，示例如下：

批量读取（按行分组） #

def read_large_file_in_chunks(file_path, chunk_size=5000):
    with open(file_path, 'r') as f:
        while True:
            # 一次性读取指定行数，返回列表
            chunk = list(itertools.islice(f, chunk_size))
            if not chunk:
                break
            # 可在此添加数据预处理逻辑（如清洗、格式转换）
            yield chunk

# 使用示例
for chunk in read_large_file_in_chunks('big_source_file.txt', 10000):
    # 将chunk交给进程池处理或写入文件
    pass

逐行读取（更细粒度） #

如果需要逐行处理，文件对象本身就是迭代器，可直接封装为生成器：

def read_large_file_line_by_line(file_path):
    with open(file_path, 'r') as f:
        for line in f:
            yield line.strip()  # 去除换行符等预处理

四、额外优化建议 #

• 若目标是写入同一个文件，不要用多进程并行写入，会导致内容混乱，此时多进程适合做数据预处理，单进程负责最终写入。
• 根据单条数据的大小调整chunk_size：数据量大则减小chunk，数据量小则增大chunk，平衡内存占用与进程调度开销。
• 可使用pool.map替代apply_async，代码更简洁：将chunk与文件名打包为任务列表，批量提交给进程池。

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