DSL2 equivalent of from (process with input channel)

[maxbates]

The docs for DSL2 include examples like:

workflow my_pipeline {
    foo()
    bar( foo.out.collect() )
}

and

workflow {
    data = channel.fromPath('/some/data/*.txt')
    foo(data)
    bar(data)
}

But is this actually possible in DSL2?

--

I am trying to write a split-apply-combine wrapper which:

• split_tarball takes a tarball with many files, and chunks it into multiple directories
• runs a sub_workflow on each chunk directory
• combine_directory_contents combines the results into a single output directory

It might look something like this:

split_tarball("in.tar.gz") | sub_workflow | flatten | collect | combine_directory_contents

I'm unclear how feed the several directories into a process like combine_directory_contents which can join all of them together.

The closest thing I see is the collectFile operator, but this only works for text.

Thanks for any help!

[GDCCP]

I have used fromFilePairs and a few other channel factory methods with DSL2. I don't think there is any additional limitations in DSL2 with respect to factory methods.

I would use collect instead of collectFile. Note that this may just be one possible implementation.

I am thinking having a process/workflow SubW that output a directory (nextflow path output does recusively copy everything if given a directory). So, feeding SubW.out.collect() to combine_direcctory_contents will get close to what you need.

Experiment by just viewing() this channel. You may need to flatten/sort as you suggested.

[maxbates]

This doesn't really get at the meat of the question, which is how to implement a combine_directory_contents that takes an arbitrary number of arguments in DSL2, not collect the results.

It's straightforward to view the results as they are streamed in, but I am not clear how to aggregate the stream into a single value, whether it's with a process or operator or something else.

[GDCCP]

So you will have a process combine_directory_contents. When it is triggered, all the collected output(directories) is available. How you want to combine these into a new output/directory as you originally asked can be implemented within script: of the process, I suppose. You can use shell commands plus groovy/java code and DSL2 features to manipulate the collected value (which can be an array of any structure, but in this context, a array of directories).

say dirarray is the input name and dest is the output directory name, you may do

process.sh ${dest} ${dirarray.join(' ')}

and do whatever you want with the content in the directories inside process.sh. I am not sure .join(' ') satisfies your need to aggregate the 'stream' into one value. Essentially, collect() aggregate the stream into an array(possibly view as a single value), and join() combines the elements in an arrays.

[multimeric]

I'm unclear how feed the several directories into a process like combine_directory_contents which can join all of them together.

You've already got the first part right, which is using collect() to convert from a channel of paths to a channel with a list of paths. Then you can just pass in this channel to a process with the path directive (yes, even though it's a list of paths, and not a single path!), wrangle the files into a string, and use them in the CLI.

Here's a simple example process that simply merges every file in every input directory using cat:

nextflow.enable.dsl=2

process merge_files {
    echo true
	input:
		path files
	output:
		path "combined.txt"
	script:
		def combined = files.collect { it + '/*' }.join(' ')
		"""
		cat ${combined} > combined.txt
		"""
}

workflow {
	in = Channel.fromPath(params.merge, type: "dir") | collect | view
	merge_files(in) | view
}

Note that I have access to the files variable inside the script block, and I can write any arbitrary groovy code here. files is a groovy list, so I used the List.collect(closure) method here. This is not a nextflow operator.

Then, I create some simple test files to concatenate:

$ tree inputs
inputs
├── a
│   └── a.txt
└── b
    └── b.txt
$ head inputs/**.txt
==> inputs/a/a.txt <==
abc

==> inputs/b/b.txt <==
def

Finally, I run the pipeline:

$ nextflow run test.nf --merge 'inputs/*' 
N E X T F L O W  ~  version 21.04.0
Launching `test.nf` [backstabbing_carlsson] - revision: 5d6dd661a2
executor >  local (1)
[f0/1c67b1] process > merge_files [100%] 1 of 1 ✔
[/tmp/tmp.t526C2QGg7/inputs/b, /tmp/tmp.t526C2QGg7/inputs/a]
/tmp/tmp.t526C2QGg7/work/f0/1c67b1eb45be47017a1c8ce773de0d/combined.txt

And the results are:

$ cat /tmp/tmp.t526C2QGg7/work/f0/1c67b1eb45be47017a1c8ce773de0d/combined.txt                                                                         
def
abc

[maxbates]

Thanks, this is really helpful, I'll follow up once I give it a try!