Collecting multiple directories with the same name in one process

[Lehmann-Fabian]

Hello there, I have a workflow, which follows this structure:

elements = Channel.from("a","b","c").flatten()

def extractDirectory = { it.parent.toString().substring(it.parent.toString().lastIndexOf('/') + 1 ) }

process firstProcess {

    input: 
    val value from elements
    
    output:
    file ("**") into filesInDir
    
    """
    mkdir t u v
    touch t/${value}.txt
    touch u/${value}.txt
    touch v/${value}.txt
    ls
    """

}

filesByDir = filesInDir.flatten().map{ [extractDirectory(it), it] }.buffer( size: Integer.MAX_VALUE, remainder: true )

process secondProcess {

    echo true

    input: 
    tuple val(folderName), file("$folderName/*") from filesByDir
    
    """
    find . -name "*"
    """

}

In the secondProcess, I need to stage all the files, and preserve the directory structure. This process should be a single tasks that collects all data from before. In this example, following files should be staged:

t/a.txt
t/b.txt
t/c.txt
u/a.txt
u/b.txt
u/c.txt
v/a.txt
v/b.txt
v/c.txt

With another variant (see below), I get the following error:

Process secondProcess input file name collision -- There are multiple input files for each of the following file names: local-dir/v, local-dir/t, local-dir/u

elements = Channel.from("a","b","c").flatten()

def extractDirectory = { it.parent.toString().substring(it.parent.toString().lastIndexOf('/') + 1 ) }

process firstProcess {

    input:
    val value from elements
   
    output:
    file ("{t,u,v}") into filesInDir
   
    """
    mkdir t u v
    touch t/${value}.txt
    touch u/${value}.txt
    touch v/${value}.txt
    ls
    """
}


process secondProcess {
    echo true
    input:
    file("local-dir/*") from filesInDir.flatten().collect()
   
    """
    find local-dir -name "*"
    """
}

I’m really looking forward for any suggestions.
Thank you in advance!

[pditommaso]

So basically you want to stage all files in the u together in the u directory, same for the other. The secondProcess should be executed once collecting all previous outputs?

[pditommaso]

That is possible with the following:

elements = Channel.from("a","b","c").flatten()

def extractDirectory = { it.parent.toString().substring(it.parent.toString().lastIndexOf('/') + 1 ) }

process firstProcess {

    input:
    val value from elements
   
    output:
    file ("{t,u,v}/*.txt") into filesInDir
   
    """
    mkdir t u v
    touch t/${value}.txt
    touch u/${value}.txt
    touch v/${value}.txt
    ls
    """
}


filesGroupped = filesInDir
    .flatten()
    .map { file-> [file.parent.name, file] }
    .groupTuple()


process secondProcess {
    echo true
    input:
    tuple val(id), path("$id/*") from filesGroupped
   
    """
    find $id -name "*"
    """
}

The problem is that it will run a task for each group, instead, my understanding is that you need to process all together

[Lehmann-Fabian]

You are right, this is exactly what I need, but in a single process.
I will continue to rename the files in the first process, and rename them again in the second one.
Thank you very much.

[ysliaw01]

You can just zip/tar the directory, and pass the zip/tar file as output to your 2nd process. Then unzip/untar the zip/tar file.

[Lehmann-Fabian]

Thank you, @ysliaw01, for your feedback. Unfortunately tar/zip is not an option as it would generate a lot of I/O - one directory contains multiple gigabytes of data.

[Lehmann-Fabian]

I proposed a solution for the named issue here: #2622