Running a workflow on each item emitted by a channel

[kwcurrin]

Hello,

I am trying to run my pipeline on each line of a .csv file separately. I understand how to pass eachline of a .csv file to a process by splitting the csv file using the .splitCsv operator and then passing this channel to a process. However, I can't figure out how to do the same for a subworkflow. The subworkflow takes in the entire channel at once and I can't figure out how to split into the individual lines. I tried adding a process at the start of my workflow that just emits the items that I want so that I can access them as p1.out[0], etc, but this seems to result in the same issue. The subworkflow "tale" block doesn't seem to have the same tuple splitting functionality as the process "input:" block. Ultimately, what I am hoping to accomplish is the equivalent of using a bash or python for loop to run my pipeline on each sample in a csv file separately. It would be great to be able to have this functionality in nextflow directly though.

Thanks!

Kevin

[mjhipp]

I have done something similar where I used Channel.fromList() as input to the inner workflow. Each item in the list was run in parallel. I had a list of file paths, something like this:

sample_files = [
  '/local/path/sample1.bam',
  's3://bucket/sample2.bam'
]

workflow SINGLESAMPLE {
  take:
    bam_file

  main:
    PROCESS(bam)
}

workflow {
  SINGLESAMPLE(
    Channel.fromList(sample_files)
  )
}

[mjhipp]

possibly you could use something like

Channel.fromList(file.splitCsv)

[kwcurrin]

Thanks! That looks like a good option