Snakemake checkpoint未评估，BaseCellCounter缺失输入文件问题排查

我用checkpoint SplitBam生成一批BAM文件，需要交给rule BaseCellCounter处理，但Snakemake始终无法识别bam="SplitBam/{scDNA}/{scDNA}.{clone}.bam"是由该checkpoint生成的，即使加了rule AggregateSplitBamOutput也没用，DAG构建失败。

运行时get_all_splitbam_file_names函数只打印了before checkpoint，没打印after checkpoint，报错如下：

Building DAG of jobs...
before checkpoint
MissingInputException in rule BaseCellCounter in file SnakeFile.smk, line 96:
Missing input files for rule BaseCellCounter:
output: BaseCellCounter/Sample1/Sample_1.Clone_1.tsv,
wildcards: scDNA=Sample_1, clone=Clone_1
affected files:
SplitBam/Sample_1/Sample_1.Clone_1.bam

我的代码如下：

all_clones_all_samples = {'Sample_1': ['clone_1'], 'Sample_2': ['clone_1', 'clone_2']}
SAMPLES = ['Sample_1', 'Sample_2']

def get_all_splitbam_file_names(wildcards):
    print('before checkpoint')
    split_dir = checkpoints.SplitBam.get(**wildcards).output["split"]
    print('after checkpoint')
    CLONES = glob_wildcards(f'{split_dir}/{wildcards.scDNA}.{{clone}}.bam').clone
    return expand(f"{split_dir}/{wildcards.scDNA}.{{clone}}.bam",clone=CLONES)

rule all:
    input:
        expand("MergeCounts/{scDNA}.BaseCellCounts.AllCellTypes.tsv", 
        scDNA = SAMPLES),
    default_target: True

checkpoint SplitBam:
    input:
        bam = f"{DATA}/{{scDNA}}_scDNA.bam",
    output:
        split = directory("SplitBam/{scDNA}")
    shell:
        "mkdir -p {output.split} && python create_some_files.py {input.bam}"

rule AggregateSplitBamOutput:
    input:
        bam=get_all_splitbam_file_names,
    output:
        touch("SplitBam/{scDNA}.split_done.txt")

rule BaseCellCounter:
    input:
        txt="SplitBam/{scDNA}.split_done.txt",
        bam="SplitBam/{scDNA}/{scDNA}.{clone}.bam",
    output:
        tsv="BaseCellCounter/{scDNA}/{scDNA}.{clone}.tsv",

def MergeCountsInput(wildcards):
    return expand(f"BaseCellCounter/{{scDNA}}/{{scDNA}}.{clone}.tsv", 
            clone=all_clones_all_samples[wildcards.scDNA])

rule MergeCounts:
    input:
        MergeCountsInput,
    output:
        tsv = "MergeCounts/{scDNA}.BaseCellCounts.AllCellTypes.tsv"

错误原因 #

核心问题是DAG构建阶段提前触发了BaseCellCounter的依赖检查，此时checkpoint尚未运行，动态生成的BAM文件不存在，同时AggregateSplitBamOutput的设计未起到衔接作用：

1. MergeCountsInput直接用硬编码的all_clones_all_samples生成BaseCellCounter的输出路径，Snakemake会提前解析这些路径并检查对应的输入（BAM文件），但此时checkpoint还未执行，BAM文件未生成，因此报错。
2. get_all_splitbam_file_names中的checkpoints.SplitBam.get(**wildcards)在DAG构建阶段无法获取checkpoint的输出（checkpoint还未运行），导致函数卡在该步骤，无法打印after checkpoint。

解决方案 #

需要让BaseCellCounter的触发完全依赖checkpoint的动态输出，而非提前使用硬编码的克隆列表，调整步骤如下：

步骤1：修改MergeCounts的输入函数，从checkpoint获取实际生成的文件 #

替换原MergeCountsInput，改为从checkpoint输出目录动态获取TSV文件路径：

def MergeCountsInput(wildcards):
    # 获取checkpoint的输出目录
    split_dir = checkpoints.SplitBam.get(**wildcards).output["split"]
    # 从SplitBam目录提取实际生成的克隆名
    clones = glob_wildcards(f'{split_dir}/{wildcards.scDNA}.{{clone}}.bam').clone
    # 生成对应BaseCellCounter的输出路径
    return expand("BaseCellCounter/{scDNA}/{scDNA}.{clone}.tsv", scDNA=wildcards.scDNA, clone=clones)

步骤2：移除冗余的AggregateSplitBamOutput规则 #

该规则无实际作用，可直接让BaseCellCounter的输入关联checkpoint的动态输出，无需额外触发文件。

步骤3：修改BaseCellCounter的输入，通过checkpoint动态生成BAM路径 #

为BaseCellCounter添加输入函数，直接从checkpoint获取BAM文件路径：

def get_basecellcounter_input(wildcards):
    split_dir = checkpoints.SplitBam.get(**wildcards).output["split"]
    return {"bam": f"{split_dir}/{wildcards.scDNA}.{wildcards.clone}.bam"}

rule BaseCellCounter:
    input:
        get_basecellcounter_input,
    output:
        tsv="BaseCellCounter/{scDNA}/{scDNA}.{clone}.tsv",
    # 添加你的处理命令，例如：
    # shell: "your_process_command {input.bam} > {output.tsv}"

完整修改后代码 #

all_clones_all_samples = {'Sample_1': ['clone_1'], 'Sample_2': ['clone_1', 'clone_2']}
SAMPLES = ['Sample_1', 'Sample_2']

rule all:
    input:
        expand("MergeCounts/{scDNA}.BaseCellCounts.AllCellTypes.tsv", 
        scDNA = SAMPLES),
    default_target: True

checkpoint SplitBam:
    input:
        bam = f"{DATA}/{{scDNA}}_scDNA.bam",
    output:
        split = directory("SplitBam/{scDNA}")
    shell:
        "mkdir -p {output.split} && python create_some_files.py {input.bam}"

def get_basecellcounter_input(wildcards):
    split_dir = checkpoints.SplitBam.get(**wildcards).output["split"]
    return {"bam": f"{split_dir}/{wildcards.scDNA}.{wildcards.clone}.bam"}

rule BaseCellCounter:
    input:
        get_basecellcounter_input,
    output:
        tsv="BaseCellCounter/{scDNA}/{scDNA}.{clone}.tsv",
    # 补充你的处理命令

def MergeCountsInput(wildcards):
    split_dir = checkpoints.SplitBam.get(**wildcards).output["split"]
    clones = glob_wildcards(f'{split_dir}/{wildcards.scDNA}.{{clone}}.bam').clone
    return expand("BaseCellCounter/{scDNA}/{scDNA}.{clone}.tsv", scDNA=wildcards.scDNA, clone=clones)

rule MergeCounts:
    input:
        MergeCountsInput,
    output:
        tsv = "MergeCounts/{scDNA}.BaseCellCounts.AllCellTypes.tsv"
    # 补充你的合并命令

原理说明 #

• Snakemake的checkpoint机制需要在规则输入函数中延迟解析依赖，即只有当checkpoint执行完成后，才会调用输入函数获取动态生成的文件路径。
• 移除硬编码的克隆列表依赖，改为从checkpoint的实际输出中获取文件，确保DAG构建是动态的，不会提前检查不存在的文件。

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