How to use publishDir on a workflow output?

[marcadella]

It is possible to use publishDir in a process, but it doesn't seem possible to publish a workflow output channel (DSL2).

[marcadella]

Why has workflow publish feature been depreciated? That was exactly what I needed.

[elowy01]

The same issue here

[aegaskin]

Commenting to keep updated on this!

[aegaskin]

I've just learned that publishDir ${task.process}, pattern: "", mode: 'copy' publishes named process outputs, with the directory structure workflow:subworkflow:process - but I'm open to suggestions and improvements!

[cjw85]

I too am wondering this. (See also #1540)

After 1bd7b4a its not possible to use publish::

cause: Workflow 'publish' is not supported anymore use process 'publishDir' instead

Its not clear to me how to use publishDir without writing another process dedicated to output; that seems wasteful as it good lead to extra cluster jobs, starting containers, etc. etc..

[cjw85]

@pditommaso

I've found comments in gitter from Apr 2020 stating that creating a publishDir in process is the only way to publish files for now, but that "the plan is to have a more flexible mechanisms later".

Are there any timelines on that?

[mjhipp]

I am also interested in this feature. My use case is that when using DSL2 I would like my process to create temp files and my workflow to publish them to their final destination. This way I do not need to pass the outputDir down through all my sub-workflows to all my child processes.

[junjun-zhang]

We were wondering about the same when this change became first known to us. I felt that the change was counter-intuitive, why let processes decide where and whether to output files? Shouldn't the decision be at the workflow level? With some exprimenting, we settled with a solution that worked well. To some degree it achieves the effect to let the workflow to control where and whether a process should publish its output. Below is the pseudo code:

process A in a.nf

params.publish_dir = ""

process A {
  publishDir "${params.publish_dir}/${task.process.replaceAll(':', '_')}", enabled: params.publish_dir

}

process B in b.nf

params.publish_dir = ""

process B {
  publishDir "${params.publish_dir}/${task.process.replaceAll(':', '_')}", enabled: params.publish_dir

}

Workflow workflow.nf imports process A and B. It set the workflow output to workflow_outdir, it disabled output of process A but enabled output of process B.

nextflow.enable.dsl = 2

params.publish_dir = "workflow_outdir"

include { A } from "./a.nf" params([*:params, "publish_dir": ""])  // sets publish_dir to empty str for process A, effectively disabling publish_dir
include { B } from "./b.nf"   // this inexplicitly pass all params to B, ie, params(params) which sets publish_dir to 'workflow_outdir' for process B

workflow {
    A()
    B()
}

[mjhipp]

Thanks for the suggestion! I think I will use this in my processes. Also I am new to groovy, but for your enabled boolean, can you simplify to:

enabled: params.publish_dir

as the empty string will be false and non-empty string is true?

[junjun-zhang]

you are right, updated

[mjhipp]

My suggestion of simplifying the statement (to rely on groovy truth) no longer works after 21.07.0-edge. Must now have enabled: params.publish_dir as boolean if you want to get this effect.

651835d

[JAMKuttan]

+1 on this issue thread!

[keiranmraine]

My 2p is that workflows should be publishing from the workflow.onComplete block. This allows the workflow to be composed using the emit directives without impacting the original use case.

[cjw85]

@keiranmraine Is this possible now? Do you have code that can be written in a workflow.onComplete to publish files?

Or are you making a proposal for how it could be implemented? In a sense the old publishDir directive implemented a standard callback for workflows and processes.

[keiranmraine]

Sorry, I should have been clearer. I don't have a working solution, it's completely an idea of how it could be separated out from the primary workflow to ensure workflows can be composed without mixing in output locations to the processes or workflow definitions:

workflow composable_flow {
    //...
    emit:
    pindel_vcf =  tool_pindel_pl.out.vcf
    pindel_vcf_flagged =  tool_flagvcf_pl.out.flagged
}

workflow.onComplete {
    publishComplete path: params.publish_dir
    // where all files defined in the emit of the most recently triggered workflow
    // are writen to the specified path
}

Propose publishComplete as this will need different options to publishDir

[cjw85]

OK, yes I'm with you. I think I expressed before that it should be the responsibility of the highest level of scripting to be responsible for publishing and processes and workflows should simply function as objects emitting named channels, to provide separation of concerns and better modularity.

[bentsherman]

The only problem with this approach is that the onComplete hook is triggered after all files have been published, so that you have an opportunity to perform any tasks that depend on the publishing being done.

See #3107 (reply in thread) for related discussion.

[bentsherman]

Folks, I whipped up a little PR for a publish operator: #3724

Check out the pipeline example and let me know if you have any feedback.

[BJWiley233]

We are definitely getting beat out by WDL and cromwell if we cannot upload workflow outputs to a directory. +1 for this ask as a necessity to be a successful (cough: workflow :cough) language. Who can I speak with or call at Seqera to get this facilitated ASAP??

[bentsherman]

I'm wondering if we should give the people what they want until we deliver that generic solution, since the publish operator would have some nice features that copyTo() does not, like pattern matching, content type, tags, etc.

[BJWiley233]

Can you give an example of how to use subscribe + copyTo in the workflow. Below gives me the error: Missing process or function subscribe([Script_d83af895$_runScript_closure9$_closure34$_closure35@ef220e4])

params.sample = "./"

workflow {

    WF(ref, input)
    FILTER(WF.out.vcf, WF.out.vcf_index)
    
    files = [FILTER.out.vcf, FILTER.out.vcf_index, WF.out.vcf, WF.out.vcf_index]

    files.flatten().subscribe { f -> 
			f.copyTo("${params.sample}")
    }

}

Sorry but I am coming from WDL where we have file arrays Array[File] files so if NF doesn't do this then let me know.

[bentsherman]

I think you want concat:

files = FILTER.out.vcf.concat(FILTER.out.vcf_index, WF.out.vcf, WF.out.vcf_index)

[BJWiley233]

sorry I picked bad names for subworkflows: WF and FILTER are subworkflows and each emit say 2 files for now, WF emits files vcf and vcf_index and so does FILTER. How so I get these 4 files together in an array?

[BJWiley233]

oh I see, you need a file first and concat the rest 😄

[jmonroynieto]

Migrating to DSL2, I found this issue as well. I have to be very consistent about my own modules but those maintained by other people have to be copied and modified. I wanted to give @bentsherman's PR a try, but I can't figure out how to build from the source after cloning the changes. Any tips?

This would solve some headaches for me but more importantly this erodes modularity and reusability which are the selling points of DSL2.

[bentsherman]

Run make compile, then use ./launch.sh as a drop-in replacement for nextflow.