如何在不加载全量数据下并行处理双端测序文件的4行读段

针对40GB级别的配对测序数据（每组读段4行），要在不加载全量数据到内存的前提下实现并行处理，核心思路是流式分块+多进程任务队列，既保证配对读段的同步性，又控制内存占用。

核心实现步骤 #

• 流式分块读取：从两个文件中同步逐组读取4行读段，积累到指定数量后形成一个数据块，避免一次性加载全量数据。
• 多进程异步处理：用进程池异步处理每个数据块，通过流式传递数据块控制内存占用。
• 严格同步配对读段：每个数据块内的读段组都是从两个文件对应位置读取的，天然保证配对关系。

代码示例 #

import multiprocessing

# 你的原有处理函数，输入为两组各4行的读段
def process_read_pairs(data1_lines, data2_lines):
    # 此处替换为你的实际处理逻辑
    return f"Processed pair: {data1_lines[0][:10]}... <-> {data2_lines[0][:10]}..."

def process_chunk(chunk):
    """处理单个数据块，chunk内是一一对应的读段组"""
    results = []
    for d1_lines, d2_lines in chunk:
        res = process_read_pairs(d1_lines, d2_lines)
        results.append(res)
    return results

def stream_chunks(file1, file2, chunk_size=1000):
    """流式生成数据块，每次读取chunk_size组配对读段"""
    while True:
        current_chunk = []
        for _ in range(chunk_size):
            d1 = []
            for _ in range(4):
                line = file1.readline()
                if not line:
                    break
                d1.append(line)
            d2 = []
            for _ in range(4):
                line = file2.readline()
                if not line:
                    break
                d2.append(line)
            # 如果其中一个文件读完，终止当前块的积累
            if len(d1) <4 or len(d2)<4:
                break
            current_chunk.append((d1, d2))
        if not current_chunk:
            break
        yield current_chunk

def main():
    with open("data1.txt", "r") as f1, open("data2.txt", "r") as f2:
        # 根据CPU核心数设置进程数，平衡并行效率和内存占用
        pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
        chunk_generator = stream_chunks(f1, f2, chunk_size=2000)
        
        # imap_unordered允许乱序输出，速度更快；需要顺序输出则改用imap
        for processed_results in pool.imap_unordered(process_chunk, chunk_generator):
            # 此处替换为你的结果输出逻辑（如写入文件）
            for res in processed_results:
                print(res)
        
        pool.close()
        pool.join()

if __name__ == "__main__":
    main()

关键注意事项 #

• chunk_size调优：建议根据可用内存调整，1000-5000组是比较稳妥的范围，太大易导致内存占用过高，太小会增加进程间通信开销。
• 异常处理：实际部署时可添加文件读取异常捕获，避免因文件损坏或格式错误导致程序崩溃。
• 输出顺序：若允许结果乱序，用imap_unordered速度最优；若需要保持原始读段顺序，改用imap即可，性能仍远优于单进程。
• 资源监控：运行时可通过系统工具（如top、htop）监控内存和CPU使用率，动态调整进程数和chunk_size。

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