Python多进程并行处理大规模数据：共享输入文件与结果数组的可行性及安全性问询

我之前刚好处理过类似的大规模数据并行计算场景，你的需求完全可行，下面给你具体的解决方案和踩过的坑：

1. 并行读取文件的正确姿势 #

直接让多个进程同时啃同一个大文件确实会出问题——要么重复读同一行，要么把一行拆成两半读取，完全乱套。更稳妥的是两种思路：

• 先拆分大文件：把150万条记录分成N个小文件（N建议等于你的CPU核心数，比如8核就拆8份），每个进程单独处理一个小文件，彻底避免读取冲突。Linux/macOS可以直接用split命令，Windows或者想自定义拆分规则的话，写个10行以内的Python脚本就能搞定。
• 按文件偏移量分配读取区间：如果不想生成中间文件，可以先获取文件总大小，给每个进程分配起始和结束的字节偏移量，进程从指定位置开始读，注意要跳过开头不完整的行，保证每次读的都是完整记录。这种方式稍复杂，但适合不想额外占存储的场景。

2. 保证计算结果的准确性 #

你担心多个进程同时累加同一个result数组会有竞争问题，这确实是个坑——多进程的内存是相互隔离的，直接共享numpy数组要么导致数据错乱，要么直接报错。正确的做法是：

• 每个进程初始化自己的局部结果数组（和全局result同形状的np.zeros((1000,1000))）
• 所有进程计算完成后，把所有局部结果数组汇总相加，得到最终的全局结果

这种方式完全避免了进程间的资源竞争，结果100%准确。

3. 具体代码实现示例 #

方案一：拆分文件后并行处理 #

import numpy as np
from multiprocessing import Pool

def process_single_chunk(chunk_file):
    # 每个进程独立维护自己的局部结果
    local_result = np.zeros(shape=(1000,1000))
    with open(chunk_file, "r") as file:
        for line in file:
            n = calculateThings(float(line))
            local_result += n
    return local_result

def calculateThings(data):
    # 保留你原本的计算逻辑
    return np.where(some_border_condition, function(data), 0)

if __name__ == "__main__":
    # 假设已经把大文件拆成了8个小文件
    chunk_files = [f"input_chunk_{i}.txt" for i in range(8)]
    
    # 创建进程池，进程数和CPU核心数匹配最佳
    with Pool(processes=8) as pool:
        # 并行处理所有文件块
        all_local_results = pool.map(process_single_chunk, chunk_files)
    
    # 汇总所有局部结果得到最终值
    final_result = sum(all_local_results)
    # 后续可以保存或处理final_result

方案二：按偏移量读取（不拆分文件） #

import numpy as np
from multiprocessing import Pool

def process_file_range(args):
    file_path, start_offset, end_offset = args
    local_result = np.zeros(shape=(1000,1000))
    with open(file_path, "r") as file:
        file.seek(start_offset)
        # 跳过开头不完整的行（如果不是文件起始位置）
        if start_offset != 0:
            file.readline()
        # 读取到分配的结束位置
        while file.tell() < end_offset:
            line = file.readline()
            if not line:
                break
            n = calculateThings(float(line))
            local_result += n
    return local_result

def calculateThings(data):
    return np.where(some_border_condition, function(data), 0)

if __name__ == "__main__":
    file_path = "inputfile.txt"
    # 获取文件总大小
    with open(file_path, "r") as f:
        f.seek(0, 2)
        total_size = f.tell()
    
    num_processes = 8
    chunk_size = total_size // num_processes
    # 给每个进程分配读取区间
    process_args = []
    for i in range(num_processes):
        start = i * chunk_size
        # 最后一个进程读到文件末尾
        end = (i+1)*chunk_size if i != num_processes-1 else total_size
        process_args.append((file_path, start, end))
    
    with Pool(processes=num_processes) as pool:
        all_local_results = pool.map(process_file_range, process_args)
    
    final_result = sum(all_local_results)

4. 额外优化小技巧 #

• 拆分文件时尽量保证每个小文件的行数差不多，避免有的进程早早干完，有的还在硬扛
• 如果calculateThings里的逻辑可以向量化，优先用numpy的向量化运算代替循环，单进程速度能再提一截
• 不需要保持结果顺序的话，可以用pool.imap_unordered代替map，能稍微提升一点效率

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