Nextflow: Can I parameterize similar processes inside a for loop?

Question

I have set of ~5 processes which perform very similar tasks such that their differences can be parameterized. Rather than putting 5 items through some channel (which seems complicated), or defining 5 separate processes, it appears to be possible in Nextflow to declare the processes in a for loop, and output into an array of channels.

Below is a prototypical example and it seems to work as intended. (I understand this could be done differently with some channel cross product operator but I am just giving this as an example of a process in a for loop.)

Is there anything wrong with doing this? Is there a better way to do it?

Bonus question - the name of the process below is "foo". I would like to have the index reflected in the name i.e. "foo_$i" but this doesn't seem possible by default - any way to do this??

some_inputs = ['prot','dna', 'rna']

some_parameters = [5,3,2,7,1]

// Set up an array of channels to output to
receivers = new ArrayList<Channel>()
for (i = 0; i < 5; i++) {
    receivers.add(Channel.create())
}

// Iterate 5 times to create 5 processes each with different parameters
for (i = 0; i < 5; i++) {

    // WOW - process can be defined inside a for loop!!
    process "foo" {

      // Extract parameters here so correct $i value is used
      p = some_parameters[i]

      input:
      val x from Channel.from(some_inputs)

      output:
      val "$x $p" into receivers[i] // Note that correct index $i seems to be used here

      """
      """
    }
}

// Confirm that each of the processes outputted independently with the correct parameter
receivers[0].view { "Received[0]: $it" }
receivers[1].view { "Received[1]: $it" }
receivers[2].view { "Received[2]: $it" }
receivers[3].view { "Received[3]: $it" }
receivers[4].view { "Received[4]: $it" }

Output - note that the parameter is consistent within each input channel i.e. working correctly:

N E X T F L O W  ~  version 20.10.0
Launching `play.nf` [fabulous_keller] - revision: c89e2b2281
executor >  local (15)
[10/b477ec] process > foo (3) [100%] 3 of 3 ✔
[16/8282ba] process > foo (2) [100%] 3 of 3 ✔
[db/f7327e] process > foo (1) [100%] 3 of 3 ✔
[fb/886f6f] process > foo (1) [100%] 3 of 3 ✔
[46/c34ed3] process > foo (1) [100%] 3 of 3 ✔
Received[1]: rna 3
Received[1]: prot 3
Received[4]: rna 1
Received[2]: rna 2
Received[0]: dna 5
Received[2]: dna 2
Received[2]: prot 2
Received[0]: prot 5
Received[3]: rna 7
Received[0]: rna 5
Received[1]: dna 3
Received[3]: dna 7
Received[4]: dna 1
Received[3]: prot 7
Received[4]: prot 1

Answer 1

Use channels, this is what nextflow's entire design is based on. I'm honestly surprised that spawning 5 processes with the same name in a loop even works.

Turn your parameters into regular channels, use the combine operator (https://www.nextflow.io/docs/latest/operator.html#combine) to transform the 2 parameter channels into a channel that has each combination of params/inputs. Feed this channel into the foo process. Afterward use branch (https://www.nextflow.io/docs/latest/operator.html#branch) to split your foo output channel into 5 different channels.

typechan = Channel.from(5,3,2,7,1)
numchan = Channel.from('prot','dna', 'rna')

cartesian = numchan.combine(typechan)

process foo {
  input:
  val numtype from cartesian
  output:
  val numtype into bar 
  script:
  """
  echo $numtype
  """
}

bar.branch {
  one: it[1] == 1
  two: it[1] == 2
  three: it[1] == 3 
  five: it[1] == 5
  seven : it[1] == 7
}.set {result}
result.one.view()
```

Answer 2

You don't need to create processes in a loop when you have channels. The each qualifier allows you to repeat the execution of a process for each item in a collection. For example, the contents of 'test.nf':

some_inputs = Channel.of( 'prot', 'dna', 'rna' )
some_parameters = Channel.of( 5, 3, 2, 7, 1 )

process foo {

    tag { "task: ${task.index} - ${x},${y}" }

    input:
    val x from some_inputs
    each y from some_parameters

    output:
    tuple val(x), val(y) into receivers

    """
    echo "x: ${x}; y: ${y}"
    """
}

results = receivers.branch { x, y ->
    
    five: y == 5
        return tuple( x, y )

    three: y == 3
        return tuple( x, y )

    two: y == 2
        return tuple( x, y )

    seven: y == 7
        return tuple( x, y )
    
    one: y == 1
        return tuple( x, y )
}

results.five.view { "My result for param 5 is: $it" }
results.three.view { "My result for param 3 is: $it" }
results.two.view { "My result for param 2 is: $it" }
results.seven.view { "My result for param 7 is: $it" }
results.one.view { "My result for param 1 is: $it" }

Then running like:

nextflow run -ansi-log false test.nf

Results:

N E X T F L O W  ~  version 20.10.0
Launching `test.nf` [fabulous_franklin] - revision: 14c3290d2b
[6e/c12ce7] Submitted process > foo (task: 7 - dna,3)
[93/616f93] Submitted process > foo (task: 4 - prot,7)
[27/0bc3c4] Submitted process > foo (task: 3 - prot,2)
[0d/0d0b51] Submitted process > foo (task: 9 - dna,7)
[d4/f56d18] Submitted process > foo (task: 6 - dna,5)
[b0/d1535f] Submitted process > foo (task: 8 - dna,2)
[13/0a660a] Submitted process > foo (task: 5 - prot,1)
[f0/1c2f57] Submitted process > foo (task: 2 - prot,3)
[3f/3e893d] Submitted process > foo (task: 1 - prot,5)
[19/2921de] Submitted process > foo (task: 10 - dna,1)
My result for param 3 is: [dna, 3]
My result for param 7 is: [prot, 7]
My result for param 7 is: [dna, 7]
[a7/953a2b] Submitted process > foo (task: 11 - rna,3)
[1c/0fd4ef] Submitted process > foo (task: 12 - rna,5)
My result for param 1 is: [prot, 1]
[39/d21adf] Submitted process > foo (task: 14 - rna,7)
My result for param 5 is: [dna, 5]
My result for param 2 is: [prot, 2]
My result for param 2 is: [dna, 2]
[f6/8509a7] Submitted process > foo (task: 15 - rna,1)
My result for param 5 is: [prot, 5]
[6f/cee0ae] Submitted process > foo (task: 13 - rna,2)
My result for param 3 is: [rna, 3]
My result for param 1 is: [dna, 1]
My result for param 2 is: [rna, 2]
My result for param 3 is: [prot, 3]
My result for param 1 is: [rna, 1]
My result for param 5 is: [rna, 5]
My result for param 7 is: [rna, 7]

The answer to your bonus question is to use the task.index variable in your script. I used this to tag your job, but it can be used anywhere in the process declaration. You likely won't need this variable though.

Usually, using the branch operator is enough for most use cases. However,it's possible to implement your own branching function. The following uses the new DSL, DSL2, to automatically fork the 'receivers' channel:

nextflow.enable.dsl=2

process foo {

    tag { "task: ${task.index} - ${x},${y}" }

    input:
    val x
    each y

    output:
    tuple val(x), val(y)

    """
    echo "x: ${x}; y: ${y}"
    """
}

def branchByElement( input_ch, elements, by=0 ) {

    def targets = [:]
    for( element in elements ) {
        def key = element
        targets[ key ] = { input_ch.filter { it[by] == key } }
    }
    return targets
}

workflow {

    some_inputs = [ 'prot', 'dna', 'rna' ]
    some_parameters = [ 5, 3, 2, 7, 1 ]

    inputs_ch = Channel.fromList( some_inputs )
    parameters_ch = Channel.fromList( some_parameters )

    foo( inputs_ch, parameters_ch )

    parameter_results = branchByElement( foo.out, some_parameters, by=1 )
    input_results = branchByElement( foo.out, some_inputs, by=0 )

    parameter_results[7].call().view { "Param 7: $it" }
    input_results.dna().view { "Input dna: $it" }
}

Results:

N E X T F L O W  ~  version 20.10.0
Launching `test.nf` [distraught_meninsky] - revision: 04df251752
[bb/32325f] Submitted process > foo (task: 4 - prot,7)
[a0/19fdf5] Submitted process > foo (task: 8 - dna,2)
[e9/4c1d09] Submitted process > foo (task: 6 - dna,5)
[a3/dcb002] Submitted process > foo (task: 2 - prot,3)
[db/7e612e] Submitted process > foo (task: 5 - prot,1)
[cd/dfa84e] Submitted process > foo (task: 9 - dna,7)
[d4/6a85f2] Submitted process > foo (task: 7 - dna,3)
[f0/8e8bf4] Submitted process > foo (task: 1 - prot,5)
[89/f20a28] Submitted process > foo (task: 3 - prot,2)
Input dna: [dna, 2]
Param 7: [prot, 7]
Input dna: [dna, 5]
[53/056fe4] Submitted process > foo (task: 10 - dna,1)
[f3/a61abf] Submitted process > foo (task: 12 - rna,3)
Input dna: [dna, 7]
[f9/a568a8] Submitted process > foo (task: 13 - rna,2)
Param 7: [dna, 7]
[06/72bb6c] Submitted process > foo (task: 11 - rna,5)
Input dna: [dna, 3]
[a3/6d80d8] Submitted process > foo (task: 14 - rna,7)
[ac/acdb5a] Submitted process > foo (task: 15 - rna,1)
Param 7: [rna, 7]
Input dna: [dna, 1]