如何在Snakemake单个Rule中高效批量处理多文件？

问题核心 #

在Snakemake中用单个Rule批量处理多文件时，遇到两个痛点：

• 单个输出文件缺失/输入更新时，会触发全量重跑，耗时过长
• 处理脚本修改后，Snakemake无法自动感知，需手动触发重跑

工作流背景 #

工作流用于筛选原始数据并输出结果，目录结构如下：

.
├── data
│   ├── 000_raw
│   │   ├── 15_a.csv
│   │   ├── 15_b.csv
│   │   ├── 15_c.csv
│   │   ├── 16_a.csv
│   │   ├── 16_b.csv
│   │   └── 16_c.csv
│   └── 010_sel
│       ├── 15_a.csv
│       ├── 15_b.csv
│       ├── 15_c.csv
│       ├── 16_a.csv
│       ├── 16_b.csv
│       └── 16_c.csv
├── scripts
│   └── 010_sel.py
└── Snakefile

筛选脚本010_sel.py单次仅处理一个文件，常规调用方式：

python scripts/010_sel.py data/000_raw/15_a.csv data/010_sel/15_a.csv

当前实现的问题 #

原有方案用expand批量生成输入输出列表，在单个Rule里循环处理：

ls_year_type = ["15_a","15_b","15_c","16_a","16_b","16_c"]

rule sel_010:
    input:
        expand("data/000_raw/{year_type}.csv", year_type=ls_year_type)
    output:
        expand("data/010_sel/{year_type}.csv", year_type=ls_year_type)
    run: 
        for ifile in range(len(output)):
            os.system("python scripts/010_sel.py {} {}".format(input[ifile],output[ifile]))

存在的问题：

• 单个输出文件缺失时，Snakemake会重新运行所有文件的处理逻辑
• 用run调用外部脚本，Snakemake无法监控脚本修改，需手动触发重跑

最优解决方案 #

核心思路是用通配符（Wildcard）将单个文件的处理逻辑封装为独立Rule，让Snakemake为每个文件生成独立任务，同时用script指令让Snakemake监控脚本变化。

1. 修改Snakefile #

# 定义所有需要处理的样本ID
SAMPLES = ["15_a", "15_b", "15_c", "16_a", "16_b", "16_c"]

# 入口规则：指定所有需要生成的最终输出文件
rule all:
    input:
        expand("data/010_sel/{sample}.csv", sample=SAMPLES)

# 单个文件处理规则：通过通配符{sample}关联输入输出
rule process_single_sample:
    input:
        raw_data = "data/000_raw/{sample}.csv"
    output:
        selected_data = "data/010_sel/{sample}.csv"
    # 使用script指令，让Snakemake监控脚本文件的修改
    script:
        "scripts/010_sel.py"

2. 修改处理脚本010_sel.py #

因为现在每个任务仅处理单个文件，无需循环，直接通过Snakemake内置对象获取输入输出：

# 从Snakemake对象获取单个输入输出文件路径
input_path = snakemake.input.raw_data
output_path = snakemake.output.selected_data

# 这里替换为你的实际筛选逻辑
import pandas as pd
df = pd.read_csv(input_path)
# 示例：筛选某列值大于0的行
filtered_df = df[df["value"] > 0]
filtered_df.to_csv(output_path, index=False)

方案优势 #

• 增量处理：删除单个输出文件（如data/010_sel/15_b.csv）后，Snakemake仅会重新处理对应样本的任务，不会全量重跑
• 脚本修改感知：用script指令后，Snakemake会监控010_sel.py的修改时间，脚本更新后自动触发所有依赖任务重跑
• 并行支持：运行时添加-j N参数（N为并行数），可同时处理多个文件，提升效率
• 扩展性强：新增样本时，仅需在SAMPLES列表中添加ID，无需修改Rule逻辑

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