Snakemake partial expand场景下如何让download规则单输入执行支持并行

解决方案 #

核心逻辑是给download规则增加tag通配符，让每个tissue_name和tag的组合对应一个独立任务，Snakemake会自动为每个输入csv生成单独的下载任务，支持并行执行。

修改点说明 #

1. 移除download规则input中的expand函数，直接用通配符匹配单个csv文件，避免一次性传入所有输入
2. 给download规则增加tag通配符，匹配每个输入文件的tag后缀
3. 改用标记文件作为download规则的输出，解决多任务写同一目录无法判断执行状态的问题
4. 调整all规则的输入，匹配所有需要的下载任务标记文件

修改后的完整Snakefile #

import csv
import os

def get_tissue_name():
    tissue_data = []
    with open("master_init.csv", "r") as rfile:
        reader = csv.reader(rfile)
        for line in reader:
            id = line[1].split("_")[0]  # naiveB_S1R1 -> naiveB
            tissue_data.append(id)
    return tissue_data

def get_tag_data():
    tag_data = []
    with open("master_init.csv", "r") as rfile:
        reader = csv.reader(rfile)
        for line in reader:
            tag = line[1].split("_")[-1]
            tag_data.append(tag)  # example: S1R1
    return tag_data

# 提前获取所有(tissue, tag)组合，供all规则使用
TISSUES = get_tissue_name()
TAGS = get_tag_data()

rule all:
    input:
        # 收集所有下载任务的完成标记
        expand("output/{tissue_name}/{tag}.download.done",
               tissue_name=TISSUES,
               tag=TAGS)

rule distribute:
    input: "master_init.csv"
    output: "init/{tissue_name}_{tag}.csv"
    params:
        id = "{tissue_name}_{tag}"
    run:
        lines = open(str(input), "r").readlines()
        wfile = open(str(output), "w")
        for line in lines:
            line = line.rstrip()
            if params.id in line:
                wfile.write(line+"\n")
        wfile.close()

rule download:
    input: "init/{tissue_name}_{tag}.csv"
    # touch会在任务完成后自动生成标记文件
    output: touch("output/{tissue_name}/{tag}.download.done")
    shell:
        """
        # 先创建输出目录避免报错
        mkdir -p $(dirname {output})
        while read srr name endtype; do
            fastq-dump --split-files --gzip $srr --outdir $(dirname {output})
        done < {input}
        """

运行效果 #

修改后执行snakemake --cores N（N为你想要的并行数），Snakemake会生成对应数量的download任务，每个任务对应一个csv输入，并行执行下载，不会再出现多输入文件的语法错误。

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