Snakemake工作流如何读取运行列表传入值数组处理批量任务

Snakemake 采用输出目标驱动的执行逻辑，和Nextflow的数据流驱动模式有本质区别，不需要专门定义输出元组的进程，只需要两步：先从运行列表解析出所有需要生成的目标文件，再定义通用规则匹配单个文件的生成逻辑即可。

核心实现逻辑 #

• 工作流启动前先解析run-list.txt，跳过注释行和空行，按照命名规则生成所有待产出的h5文件路径，作为工作流的顶层目标
• 定义入口规则rule all，将所有目标文件作为它的输入，Snakemake启动后会自动从这个规则出发，倒推每个文件的生成依赖
• 定义通用的数据拉取规则，通过通配符匹配每个输出文件名里的探测器ID、运行号，在规则内实现数据库/存储系统的查询拉取逻辑

可直接运行的代码示例 #

1. 运行列表文件（run-list.txt） #

保持现有格式即可：

# detector_id run_number
75 63433
75 67325
42 57584
42 57899
42 58998

2. Snakefile 内容 #

# --------------------------
# 第一步：解析运行列表生成所有目标文件
# --------------------------
targets = []
with open("run-list.txt", "r", encoding="utf-8") as f:
    for line in f:
        line = line.strip()
        # 跳过空行和注释行
        if not line or line.startswith("#"):
            continue
        det_id, run_num = line.split()
        # 按照要求生成固定格式文件名：探测器ID补零到8位，运行号补零到8位
        out_name = f"RUN_{int(det_id):08d}_{int(run_num):08d}.h5"
        targets.append(out_name)

# --------------------------
# 第二步：定义入口规则，指定所有要生成的目标
# --------------------------
rule all:
    input:
        targets
    # 可选：把run-list.txt本身加为依赖，列表修改时自动触发更新
    # input:
    #     targets,
    #     "run-list.txt"

# --------------------------
# 第三步：定义单条运行数据的拉取规则
# --------------------------
rule fetch_run_data:
    output:
        # 通配符匹配输出文件，加\d+约束避免非法匹配
        "RUN_{det_id:\d+}_{run_num:\d+}.h5"
    params:
        # 提取原始数值格式的ID方便调用查询接口
        raw_det = lambda wc: int(wc.det_id),
        raw_run = lambda wc: int(wc.run_num)
    shell:
        """
        # 替换成实际的数据拉取命令，比如iRODS/XRootD的拉取指令、数据库查询脚本
        echo "正在拉取探测器 {params.raw_det} 运行号 {params.raw_run} 的数据"
        # 示例：生成测试文件，实际使用时替换为拉取逻辑
        touch {output}
        """

原有伪代码存在的问题 #

• 缺少Snakemake必需的顶层入口规则rule all，引擎无法知道要生成哪些目标文件
• 直接把run-list的行内容作为规则输入是错误的：规则的input字段是指当前任务执行前必须存在的前置依赖文件，不是用来传入启动参数的入口
• shell块内直接写Python变量拆分逻辑无效，shell块内执行的是Bash命令，通配符提取到的参数需要通过wildcards对象获取
• 输出文件名的补零格式没有匹配实际需求

可选优化 #

如果习惯用Python原生逻辑写数据拉取流程，可以把shell块替换为run块，直接在规则内写Python代码：

rule fetch_run_data_py:
    output:
        "RUN_{det_id:\d+}_{run_num:\d+}.h5"
    run:
        det_id = int(wildcards.det_id)
        run_num = int(wildcards.run_num)
        # 在这里直接写Python逻辑，比如调用iRODS Python API、数据库查询驱动等
        print(f"拉取探测器 {det_id} 运行号 {run_num} 的数据")
        # 数据拉取完成后确保生成output对应的文件即可
        with open(output[0], "w") as f:
            pass

如果运行列表不是固定的，会在工作流执行过程中动态生成，可以用Snakemake的checkpoint机制实现动态目标解析，固定列表场景下直接在启动阶段解析是最简洁、性能最好的实现方式，完全符合Snakemake的设计规范。

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