A way to partially split a groupedTuple by chunks, respecting key

[VasLem]

Hi, I have the following use case but I cannot wrap my mind around how to achieve in NF, any suggestions are welcome:

USE CASE

Initial conditions: I process multiple samples at a single run. Every sample is being represented by a sample name and a variable-sized list of files, e.g, [sample1, [file1, file2,..,file15]], [sample2, [file1, file2,..,file310]] . This list is dynamically computed during runtime, by scanning a directory for files that starts with sample1, sample2, etc.
Problem: I have a process that is spawned on a SPOT machine on GCP, which can be terminated at any time, receiving a channel that emits the aforementioned sample information. I would like to pass the information in a set of operators (?) so that to receive at the end a Channel that emits, e.g, [sample1, [file1,...file10]], [sample1, [file11,...file15]], [sample2, [file1, ..., file10]].....

I guess I need something combining transpose and buffer, but I have not yet managed to get to the expected solution. This is what I have already done:

buffered = samples_and_reads | transpose 
buffered.buffer( size: 10 ).map {it -> [it[0][0], it.each{p->p[1]}]}

which clearly does not respect the groups boundaries, therefore I am getting something like [sample1, [file11,..file15,file1,..file5]] , following the example provided before. Thank you in advance!

[robsyme]

It is not quite clear what you are trying to achieve, so the suggestion below may or may not be appropriate. If I've misunderstood, would you be able to provide a small example workflow showing the problem?

Given input data:

data
├── sample1
│   ├── data.1.dat
│   ├── data.2.dat
│   ├── data.3.dat
│   ├── data.4.dat
│   ├── data.5.dat
│   └── data.6.dat
└── sample2
    ├── data.1.dat
    ├── data.2.dat
    ├── data.3.dat
    └── data.4.dat

We can generate a channel that meets your input requirements:

workflow {
    Channel.fromPath("data/sample*/*.dat")
    | map { [it.getParent().name, it] }
    | groupTuple
    | view
}

Which emits a channel with contents:

[sample2, [data/sample2/data.1.dat, data/sample2/data.3.dat, data/sample2/data.2.dat, data/sample2/data.4.dat]]
[sample1, [data/sample1/data.1.dat, data/sample1/data.3.dat, data/sample1/data.2.dat, data/sample1/data.6.dat, data/sample1/data.5.dat, data/sample1/data.4.dat]]

Now let's say that we want to split up the List of files into chunks of at most 3 files. We can use the flatMap operator like so:

workflow {
    Channel.fromPath("data/sample*/*.dat")
    | map { [it.getParent().name, it] }
    | groupTuple
    | flatMap { id, files -> files.collate(3).collect { chunk -> [id, chunk] } }
    | view
}

Which returns a channel that emits:

[sample2, [data/sample2/data.1.dat, data/sample2/data.3.dat, data/sample2/data.2.dat]]
[sample2, [data/sample2/data.4.dat]]
[sample1, [data/sample1/data.1.dat, data/sample1/data.3.dat, data/sample1/data.2.dat]]
[sample1, [data/sample1/data.6.dat, data/sample1/data.5.dat, data/sample1/data.4.dat]]

Is this close to what you were looking for?

[VasLem]

Yes, that's exactly that! So technically I should have used flatMap to process within-group tuples. Thanks for the elegant solution!

[robsyme]

Fantastic! Happy to help, Vassilis.