如何为Snakemake工作流高效分组处理样本？

要高效利用Snakemake的并行调度能力，推荐用分组作为规则的wildcard，同时保留单样本处理规则，通过依赖关系串联单样本步骤与组级步骤，具体实现如下：

1. 预处理数据表，构建分组-样本映射 #

在Snakefile顶部加载数据表，生成分组到对应样本列表的字典：

import pandas as pd

# 替换为你的数据表加载方式，比如从csv读取或直接用已有DataFrame
df = pd.read_csv("sample_info.csv")
groups = df['Group'].unique()
# 构建组到样本ID的映射
group_to_samples = {g: df[df['Group'] == g]['SampleID'].tolist() for g in groups}

2. 优化单样本处理规则 #

保留原有的单样本处理逻辑，调整取值函数确保按SampleID正确匹配：

def get_read1(sample):
    return df.loc[df['SampleID'] == sample, 'Read1'].iloc[0]
def get_read2(sample):
    return df.loc[df['SampleID'] == sample, 'Read2'].iloc[0]

rule trim_reads:
    input:
        read1=get_read1,
        read2=get_read2
    output:
        trimmed_r1="trimmed/{sample}_R1.fq.gz",
        trimmed_r2="trimmed/{sample}_R2.fq.gz"
    shell:
        """
        # 替换为你的reads修剪命令，比如cutadapt
        cutadapt -q 20 -o {output.trimmed_r1} -p {output.trimmed_r2} {input.read1} {input.read2}
        """

3. 定义组级分析规则 #

新增以group为wildcard的规则，依赖该组所有样本的处理结果，Snakemake会自动先并行完成该组所有样本的修剪，再启动组级分析：

rule analyze_group:
    input:
        # 动态获取当前组所有样本的修剪后文件
        trimmed_files = lambda wildcards: [
            f"trimmed/{s}_R1.fq.gz" for s in group_to_samples[wildcards.group]
        ] + [
            f"trimmed/{s}_R2.fq.gz" for s in group_to_samples[wildcards.group]
        ]
    output:
        group_result="group_results/group_{group}_analysis.txt"
    shell:
        """
        # 替换为你的组级分析命令，比如合并reads后做聚类、差异分析等
        cat {input.trimmed_files} | your_group_analysis_tool > {output.group_result}
        """

4. 设置全局目标规则 #

定义目标规则，让Snakemake自动处理所有分组：

rule all:
    input:
        [f"group_results/group_{g}_analysis.txt" for g in groups]

核心优势 #

• 最大化资源利用率：单样本修剪任务可并行执行，组级任务自动等待该组样本处理完成，完全发挥Snakemake的调度能力。
• 流程自动化：依赖关系由Snakemake自动管理，无需手动控制执行顺序。
• 扩展性强：新增分组或样本时，只需更新数据表，规则无需修改。

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