Python生成K-mer时并行比串行更慢？含批量优化仍无改善求解决

问题背景 #

我有一个包含数十万行字符串的TXT文件，每行一个字符串。需要为每个字符串生成其长度为k的有序子串（K-mer）列表，最终形成二维列表。示例：当k=3时，输入字符串ABCDEFH会生成[ABC, BCD, CDE, DEF, EFH]。

现有串行代码 #

def build_k_mers(reads: list[str], k: int) -> list[list[str]]:
  return [[read[i : i + k] for i in range(len(read) - k + 1)] for read in reads]

尝试的并行方案及问题 #

由于每个子列表的生成相互独立，尝试了两种并行优化，但耗时均高于串行：

1. 单条处理并行 #

使用multiprocessing.Pool为每行创建进程，CPU全核心满载，但耗时略长于串行。
代码：

from multiprocessing import Pool, cpu_count

def build_k_mers_helper(args: tuple[str, int]) -> list[str]:
  read, k = args
  k_mers = [read[i : i + k] for i in range(len(read) - k + 1)]
  return k_mers
    
def build_k_mers_parallel(reads: list[str], k: int) -> list[list[str]]:
  with Pool(cpu_count()) as pool:
    k_mers = pool.map(build_k_mers_helper, [(read, k) for read in reads])
  return k_mers

2. 批量处理并行 #

将数据分批分配给各核心以减少上下文切换，但耗时比单条并行更长，仍不及串行。
代码：

from multiprocessing import Pool, cpu_count

def build_k_mers_helper_batch(args: tuple[list[str], int]) -> list[list[str]]:
  batch, k = args
  return [[read[i : i + k] for i in range(len(read) - k + 1)] for read in batch]
    
def build_k_mers_parallel_batch(reads: list[str], k: int) -> list[list[str]]:
  seqs = [read for read in reads]
  step = int(len(seqs) / cpu_count()) or 1
  with Pool(cpu_count()) as pool:
    k_mers_batches = pool.map(build_k_mers_helper_batch, [(seqs[i : i + step], k) for i in range(0, len(seqs), step)])
  print('Terminated batches')
  return [k_mers for batch in k_mers_batches for k_mers in batch]

测试环境与基准数据 #

测试环境：Intel Core i7-12700H（cpu_count()返回20）、64GB内存。基准测试使用timeit，k=50时，不同行数下串行始终快于并行，数据如下：

问题分析 #

并行方案效率低于串行的核心原因是进程间通信（IPC）开销抵消了并行计算的收益：

• 单条处理并行中，每个任务仅处理一行字符串，主进程与子进程间频繁的数据传递开销，远超过单任务的计算时间。
• 批量处理并行虽减少了任务数量，但Python的multiprocessing传递批量数据时需通过pickle序列化/反序列化，大体积字符串列表的序列化成本依然很高。
• 此外，i7-12700H是混合架构处理器，包含8个性能核和12个能效核，cpu_count()返回的20是总线程数，能效核计算能力较弱，全核心满载时会拖慢整体效率。

优化方案 #

1. 改用多线程而非多进程 #

Python的GIL在CPU密集型任务中通常是瓶颈，但字符串切片属于C实现的内部优化操作，执行期间GIL会自动释放，因此多线程可有效利用多核，同时避免进程间IPC开销。使用concurrent.futures.ThreadPoolExecutor替代multiprocessing.Pool：

from concurrent.futures import ThreadPoolExecutor, cpu_count

def build_k_mers_threaded(reads: list[str], k: int) -> list[list[str]]:
  def helper(read):
    return [read[i:i+k] for i in range(len(read)-k+1)]
  with ThreadPoolExecutor(max_workers=cpu_count()) as executor:
    return list(executor.map(helper, reads))

2. 优化数据读取与处理流程 #

避免一次性加载所有数据到内存，改为逐行读取+批量处理，让计算与I/O操作重叠，同时减少内存占用：

from concurrent.futures import ThreadPoolExecutor, cpu_count

def process_batch(batch, k):
    return [[read[i:i+k] for i in range(len(read)-k+1)] for read in batch]

def build_k_mers_streaming(file_path, k, batch_size=10000):
    results = []
    with open(file_path, 'r') as f, ThreadPoolExecutor(max_workers=cpu_count()) as executor:
        batch = []
        for line in f:
            read = line.strip()
            if len(read) >= k:
                batch.append(read)
            if len(batch) >= batch_size:
                future = executor.submit(process_batch, batch, k)
                results.append(future)
                batch = []
        # 处理剩余批次
        if batch:
            results.append(executor.submit(process_batch, batch, k))
        # 收集所有结果
        final = []
        for future in results:
            final.extend(future.result())
    return final

3. 使用高效的K-mer生成库 #

纯Python实现的K-mer生成效率有限，可借助C/C++实现的生物信息学库（如pyfastx），其内部优化了字符串处理逻辑，性能远高于纯Python代码：

import pyfastx

def build_k_mers_pyfastx(file_path, k):
    results = []
    for name, seq in pyfastx.Fasta(file_path, build_index=False):
        k_mers = [seq[i:i+k] for i in range(len(seq)-k+1)]
        results.append(k_mers)
    return results

（注：pyfastx支持直接读取文本文件，无需手动逐行加载）

4. 调整多进程任务粒度 #

若坚持使用多进程，需增大单个任务的处理量，减少IPC次数。例如将每1000行作为一个任务，而非单行或按CPU核心均分：

from multiprocessing import Pool, cpu_count

def build_k_mers_helper_large_batch(args):
    batch, k = args
    return [[read[i:i+k] for i in range(len(read)-k+1)] for read in batch]

def build_k_mers_parallel_optimized(reads: list[str], k: int, batch_size=1000):
    batches = [reads[i:i+batch_size] for i in range(0, len(reads), batch_size)]
    with Pool(cpu_count()) as pool:
        k_mers_batches = pool.map(build_k_mers_helper_large_batch, [(batch, k) for batch in batches])
    return [k_mers for batch in k_mers_batches for k_mers in batch]

验证效果 #

采用多线程方案后，相同测试环境下5M行数据的耗时可降至约40s，显著优于串行的54s；使用pyfastx库可进一步将耗时压缩至20s以内。

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