Nextflow DSL2中如何将互斥流程的指定输出传递给下游任务

Great question! The redundancy you're trying to avoid is totally solvable in Nextflow DSL2. Since both your bcl and bcl_convert processes expose an identically-named fastq_ch output emit, we can leverage that consistency to create a clean, non-redundant workflow. Here are three effective approaches:

1. Explicit Channel Assignment (Most Readable) #

Define a single channel outside your conditional block, then assign it to the correct process's fastq_ch output based on your parameter. This keeps the logic straightforward and easy to follow:

workflow {
    runfolder_repaired = "${params.runfolder}".replaceFirst(/$/, "/")
    runfolder = Channel.fromPath(runfolder_repaired, type: 'dir')
    sample_data = Channel.fromPath(params.samplesheet, type: 'file')

    // Initialize an empty channel to hold our fastq input
    def fastq_input_ch = Channel.empty()

    if (!params.bcl_convert) {
        def bcl_results = bcl(runfolder, sample_data)
        fastq_input_ch = bcl_results.out.fastq_ch
    } else {
        def bcl_convert_results = bcl_convert(runfolder, sample_data)
        fastq_input_ch = bcl_convert_results.out.fastq_ch
    }

    // Route the unified channel to your downstream task
    fastqc(fastq_input_ch)
}

Why your first attempt failed: #

Your initial conditional assignment used def bcl_out inside each block, which created local variables only accessible within their respective if/else scopes. By declaring fastq_input_ch outside the blocks, we ensure it's accessible globally in the workflow scope.

2. Process Reference Variable (Most Concise) #

Since both processes accept the exact same input arguments and expose the required fastq_ch output, we can treat the process itself as a variable. This cuts down on boilerplate even more:

workflow {
    runfolder_repaired = "${params.runfolder}".replaceFirst(/$/, "/")
    runfolder = Channel.fromPath(runfolder_repaired, type: 'dir')
    sample_data = Channel.fromPath(params.samplesheet, type: 'file')

    // Select the process to run based on the parameter
    def selected_bcl_process = params.bcl_convert ? bcl_convert : bcl
    // Execute the selected process and capture its outputs
    def process_results = selected_bcl_process(runfolder, sample_data)
    // Pass only the required fastq_ch to downstream
    fastqc(process_results.out.fastq_ch)
}

This works because Nextflow allows you to reference processes as first-class citizens—you can assign them to variables and invoke them dynamically, just like functions.

3. Mixed Empty Channels #

If you prefer a more functional style, you can create conditional empty channels and mix them together. Since only one process will produce output (the other's channel remains empty), the mix will effectively pass only the valid fastq_ch:

workflow {
    runfolder_repaired = "${params.runfolder}".replaceFirst(/$/, "/")
    runfolder = Channel.fromPath(runfolder_repaired, type: 'dir')
    sample_data = Channel.fromPath(params.samplesheet, type: 'file')

    // Create conditional channels (one will be empty)
    def bcl_fastq = params.bcl_convert ? Channel.empty() : bcl(runfolder, sample_data).out.fastq_ch
    def bcl_convert_fastq = params.bcl_convert ? bcl_convert(runfolder, sample_data).out.fastq_ch : Channel.empty()

    // Mix the channels—only the non-empty one will contribute data
    fastqc(bcl_fastq.mix(bcl_convert_fastq))
}

Key Note: #

All these approaches rely on the fact that both bcl and bcl_convert produce a fastq_ch output with identical structure (paths to fastq.gz files). Since you've already ensured this in your process definitions, downstream tasks like fastqc will handle the input seamlessly regardless of which upstream process ran.

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