Snakemake：如何在集群/本地仅执行一次前置条件检查并终止流水线？

由于集群模式下，Snakemake的全局代码会在每个任务进程中重复执行，直接把检查逻辑放在全局代码里会导致多次检查，不符合需求。我们可以利用Snakemake的主进程判断机制，让检查只在主进程中运行一次，具体实现如下：

示例代码（Python逻辑实现） #

在你的Snakefile开头添加以下代码：

import os
from snakemake.exceptions import WorkflowError

# 仅在主进程执行前置检查
if workflow.is_master:
    target_dir = "/etc/myfolder"
    
    # 检查目录是否存在且非空
    if os.path.exists(target_dir):
        if len(os.listdir(target_dir)) > 0:
            raise WorkflowError(f"前置检查失败：目录 {target_dir} 非空，流水线终止")
    else:
        raise WorkflowError(f"前置检查失败：目录 {target_dir} 不存在，流水线终止")

关键说明： #

• workflow.is_master：这个属性仅在Snakemake的主进程中为True，集群任务的worker进程中为False，确保检查逻辑只执行一次。
• WorkflowError：这是Snakemake专属的异常类型，相比直接用sys.exit()，它能被Snakemake正确捕获，输出清晰的错误提示后终止整个流水线，更符合工具的错误处理流程。
• 可自定义检查逻辑：你可以根据实际需求修改判断条件，比如检查文件是否存在、配置参数是否合法等。

备选方案（Shell命令前置执行） #

如果你的检查逻辑用Shell就能实现，也可以通过Snakemake的--precmd命令行参数实现，在Snakemake启动前执行一次性检查：

snakemake --precmd 'if [ -n "$(ls -A /etc/myfolder)" ]; then echo "错误：/etc/myfolder目录非空" && exit 1; fi'

这种方式无需修改Snakefile，但仅适合简单的Shell可处理的检查场景。

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