如何用Python高效读取超大FASTQ文件并统计Barcode？

嘿，针对你这种超大规模FASTQ文件的Barcode统计需求，结合你手里的20核CPU和250G大内存，我给你几个分层优化的方案，从工具选型到代码实现都有，应该能把耗时从数小时压到几十分钟甚至更短：

自己写脚本很难比得上专门为生物信息场景优化的工具，这些工具都是C++/Rust实现，IO和计算效率拉满，还原生支持多线程：

• BBMap的countbarcodes工具：这是专门为大规模序列设计的Barcode计数工具，速度快到离谱。直接用命令就能搞定，完全不需要自己写逻辑：

  countbarcodes.sh in=your_large_file.fq out=barcode_counts.txt k=16 threads=20
  

  它会自动处理FASTQ的格式，提取前16碱基并统计数量，20核跑的话，处理1.25亿条序列估计十几分钟就能完成。

• SeqKit + 多线程排序：SeqKit是轻量高效的序列处理工具集，配合GNU sort的多线程功能，也能快速完成任务：

  # 如果是压缩的FASTQ，先用pigz多线程解压
  pigz -dc -p 20 your_file.fq.gz | \
    seqkit seq -s -r 1:16 | \
    sort -S 200G --parallel=20 | \
    uniq -c > barcode_counts.txt
  

  这里-s只输出序列部分，-r 1:16提取前16碱基；sort -S 200G让sort用200G内存（避免磁盘临时文件），--parallel=20启用多线程排序，效率比单线程高好几倍。

如果需要更灵活的逻辑，比如后续要加自定义处理，推荐用Python多进程+内存映射的方案，最大化利用你的硬件资源：

Python多进程+批量读取示例 #

import multiprocessing as mp
from collections import Counter

def process_chunk(lines_chunk):
    """处理一组FASTQ行，返回Barcode计数"""
    counter = Counter()
    # FASTQ是4行一组，取第2行（索引1）
    for i in range(1, len(lines_chunk), 4):
        if lines_chunk[i].strip():
            barcode = lines_chunk[i][:16]
            counter[barcode] += 1
    return counter

def merge_counters(counters_list):
    """合并多个进程的计数结果"""
    total_counter = Counter()
    for cnt in counters_list:
        total_counter.update(cnt)
    return total_counter

def count_barcodes(file_path, num_processes=18):
    # 留2个核给系统，避免资源耗尽
    with open(file_path, 'r') as f:
        # 一次性读取所有行到内存（250G内存完全够装下几十GB的FASTQ）
        all_lines = f.read().splitlines()
        # 分割成N个进程的任务块
        chunk_size = len(all_lines) // num_processes
        line_chunks = [all_lines[i:i+chunk_size] for i in range(0, len(all_lines), chunk_size)]
    
    # 启动多进程处理
    with mp.Pool(num_processes) as pool:
        results = pool.map(process_chunk, line_chunks)
    
    # 合并结果并保存
    total_counts = merge_counters(results)
    with open('barcode_counts.txt', 'w') as out_f:
        for barcode, count in total_counts.most_common():
            out_f.write(f"{barcode}\t{count}\n")

if __name__ == '__main__':
    count_barcodes('your_large_file.fq')

进阶优化：用内存映射（mmap） #

如果文件实在太大（比如单份超过100GB），可以用mmap把文件直接映射到内存，避免一次性读取的开销：

import mmap

def count_barcodes_mmap(file_path, num_processes=18):
    with open(file_path, 'r') as f:
        with mmap.mmap(f.fileno(), length=0, access=mmap.ACCESS_READ) as mm:
            all_lines = mm.read().decode().splitlines()
            # 后面的逻辑和上面一致，分割块、多进程处理、合并结果
            line_chunks = [all_lines[i:i+len(all_lines)//num_processes] for i in range(0, len(all_lines), len(all_lines)//num_processes)]
            with mp.Pool(num_processes) as pool:
                results = pool.map(process_chunk, line_chunks)
            total_counts = merge_counters(results)
            # 保存结果...

这些小调整能进一步压榨硬件性能：

• 把文件移到SSD上：HDD的读写速度是大文件处理的最大瓶颈，移到SSD能把读取速度提升10倍以上。
• 禁用文件系统atime：执行mount -o remount,noatime /your/disk（需要root权限），减少文件读取时的元数据写入开销。
• 用多线程解压压缩文件：如果你的FASTQ是.fq.gz格式，用pigz -p 20代替gzip，多线程解压速度快很多。

内容的提问来源于stack exchange，提问作者m.i.cosacak