如何用BioPython编写函数批量处理数千个FASTA文件并提速？

针对数千个FASTA文件的反向互补生成需求，以下是从单文件优化到多核并行的完整加速方案：

1. 基础单文件优化（修复原代码冗余） #

原代码重复调用reverse_complement()，且IO操作可以更高效，优化后的单文件处理函数：

from Bio import SeqIO

def process_single_fasta(input_path, output_path):
    # 用with语句自动管理文件句柄，避免手动关闭的疏漏
    with open(output_path, "w") as out_fh:
        for rec in SeqIO.parse(input_path, "fasta"):
            rc_seq = rec.seq.reverse_complement()  # 仅计算一次反向互补，避免重复运算
            # 用f-string简化字符串拼接，比+号更高效
            out_fh.write(f">rc_{rec.id}\n")
            out_fh.write(f"{rc_seq}\n")
            # 批量处理时建议注释掉print，控制台IO会大幅拖慢速度
            # print(rec.id, rc_seq)

2. 串行批量处理（遍历目录所有FASTA） #

如果不需要多核加速，先实现基础的批量遍历逻辑：

import glob
import os

def batch_process_fastas(input_dir, output_dir, suffix=".fasta"):
    # 自动创建输出目录（不存在时）
    os.makedirs(output_dir, exist_ok=True)
    # 匹配输入目录下所有指定后缀的FASTA文件
    for input_path in glob.glob(os.path.join(input_dir, f"*{suffix}")):
        filename = os.path.basename(input_path)
        output_path = os.path.join(output_dir, f"rc_{filename}")
        process_single_fasta(input_path, output_path)
        print(f"完成: {filename}")

3. 多核并行处理（核心加速方案） #

数千个文件的场景下，利用CPU多核并行处理可以将效率提升数倍，使用concurrent.futures实现：

from concurrent.futures import ProcessPoolExecutor

def parallel_batch_process(input_dir, output_dir, suffix=".fasta", max_workers=None):
    os.makedirs(output_dir, exist_ok=True)
    # 提前整理所有任务参数
    tasks = []
    for input_path in glob.glob(os.path.join(input_dir, f"*{suffix}")):
        filename = os.path.basename(input_path)
        output_path = os.path.join(output_dir, f"rc_{filename}")
        tasks.append((input_path, output_path))
    
    # 进程池自动分配任务到多核，max_workers默认取CPU核心数
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        executor.map(lambda args: process_single_fasta(*args), tasks)
    
    print("所有文件处理完成")

4. 额外实用优化建议 #

• 错误容错：给单文件处理函数添加异常捕获，避免单个文件出错导致整个批量任务中断：

  def process_single_fasta(input_path, output_path):
      try:
          with open(output_path, "w") as out_fh:
              for rec in SeqIO.parse(input_path, "fasta"):
                  rc_seq = rec.seq.reverse_complement()
                  out_fh.write(f">rc_{rec.id}\n")
                  out_fh.write(f"{rc_seq}\n")
          print(f"成功: {input_path}")
      except Exception as e:
          print(f"失败 {input_path}: {str(e)}")
  

• 大文件适配：BioPython的SeqIO.parse是迭代式读取，内存占用极低，无需额外处理大文件；
• 进度控制：批量处理时可以每处理N个文件打印一次进度，减少控制台IO开销。

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