如何针对不确定样本集编写Snakemake流程实现预处理、过滤与后续分析？

核心问题在于过滤步骤的输出是动态的，无法提前确定哪些样本会被保留。我们可以通过checkpoint（检查点）机制来实现这种动态依赖，并配合自定义函数获取过滤后的样本列表，完整流程修改如下：

修改后的完整脚本 #

import os
import glob

sample = ['A', 'B', 'C']

# 将原filter规则改为checkpoint，处理动态输出
checkpoint filter_samples:
    input:
        expand('output/pre_process/{sample}.txt', sample=sample)
    output:
        directory('output/filter')
    shell:
        """
        # 确保输出目录存在
        mkdir -p {output}
        # 模拟过滤逻辑：这里保留A、B，排除C（实际替换为你的过滤命令）
        cp output/pre_process/A.txt output/pre_process/B.txt {output}/
        """

# 自定义函数：从checkpoint输出目录提取通过过滤的样本名
def get_filtered_samples(wildcards):
    # 获取checkpoint的输出目录路径
    checkpoint_dir = checkpoints.filter_samples.get_output(wildcards)[0]
    # 遍历目录下的所有txt文件
    filtered_files = glob.glob(os.path.join(checkpoint_dir, '*.txt'))
    # 从文件名中提取样本名（如从A.txt得到A）
    return [os.path.basename(file).split('.')[0] for file in filtered_files]

rule all:
    input:
        # 预处理阶段的输出
        expand('output/pre_process/{sample}.txt', sample=sample),
        # 动态扩展过滤后样本的最终处理输出
        expand('output/data/{sample}.txt', sample=get_filtered_samples)

rule pre_process:
    output:
        'output/pre_process/{sample}.txt'
    shell:
        """
        echo "Preprocessed {wildcards.sample}" > {output}
        """

rule process:
    input:
        # 输入为过滤后的样本文件
        'output/filter/{sample}.txt'
    output:
        'output/data/{sample}.txt'
    shell:
        """
        echo "Processed {wildcards.sample}" > {output}
        """

关键修改说明 #

1. 使用checkpoint替代普通rule
   checkpoint专门用于处理输出文件数量/内容无法提前确定的步骤，Snakemake会先执行该checkpoint，再根据实际输出规划后续流程。

2. 动态获取过滤后的样本
   get_filtered_samples函数会在Snakemake生成任务依赖时被调用，实时读取checkpoint输出目录中的文件，提取出通过过滤的样本名，确保后续process规则只处理符合条件的样本。

3. 更新rule all的输入
   通过expand配合自定义函数，动态指定最终需要生成的处理后文件，让Snakemake明确整个流程的目标产物。

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