Nested scatter

[LindoNkambule]

Hi everyone,

I have two (chr1 and chr2) TSV files, each with 10 lines, and I would like to run 20 processes in parallel, how do I do this?

What I have below works for one file, how do I extend it to multiple files?

#!/usr/bin/env nextflow

nextflow.enable.dsl=2


process basicExample {
    debug true
    tag "processing ${irg}-${org}"

    input:
    tuple val(irg), val(org)

    script:
    """
    echo --lower-bound $irg --upper-bound $org
    """
}


params.inputs_path = "/path/to/my/files"

workflow {
    common_chunks_file = "${params.inputs_path}/chunks/chunks_chr1.txt"
    common_chunks_file_channel = Channel.fromPath(common_chunks_file)
                                        .splitCsv(sep:'\t', header:['index', 'chrom', 'irg', 'org'])
                                        .map{row -> tuple(row.irg, row.org)}

    for(line: common_chunks_file_channel) {
            basicExample(line)
    }
}

I have tried the following but it doesn't work

workflow {
    chroms = Channel.of(1..2)

    chroms.each { i ->
        // chunks regions files are small, we can read each file at once
        common_chunks_file = "${params.output_path}/chunks/chunks_chr${i}.txt"
        common_chunks_file_channel = Channel.fromPath(common_chunks_file)
                                                .splitCsv(sep:'\t', header:['index', 'chrom', 'irg', 'org'])
                                                .map{row -> tuple(row.irg, row.org)}

        for(line: common_chunks_file_channel) {
                basicExample(line)
        }
    }
}

Thank you

[LindoNkambule]

Nvm, this fixed it:

workflow {
    common_chunks_files = "${params.output_path}/chunks/chunks_chr*.txt"
    common_chunks_file_j = Channel.fromPath(common_chunks_files) \
                                        | splitCsv(sep:'\t', header:['index', 'chrom', 'irg', 'org']) \
                                        | map{row -> tuple(row.irg, row.org)} \
                                        | basicExample
}

[LindoNkambule]

Follow up question:

How do I extend this to work for a process that takes tuple val(irg), val(org), file(input), file(reference) instead of tuple val(irg), val(org)?

Here's a simplified Python version of what I would like to achieve:

for i in range(1, 23):
    # chrom input files
    referenceFile = f'/path/to/reference/ref_chr{i}.bcf'
    inputFile = f'/path/to/inputs/input_chr{i}.bcf'

    common_chunks_file = pd.read_csv(f'/path/to/chunks/chunks_chr{i}.txt',
                                     sep='\t', header=None,
                                     names=['index', 'chrom', 'irg', 'org'])

    # list of tuple(irg, org)
    common_chunks_regions = [(irg, org) for irg, org in zip(common_chunks_file['irg'], common_chunks_file['org'])]

    for chunk in common_chunks_regions:
        my_function(irg=chunk[0], org=chunk[1], input=inputFile, reference=referenceFile)

Thank you

[mribeirodantas]

Why can't you load your CSV files with Channel.fromPath + splitCsv, merge them and simply call the process? I can't see what's the need for all these for loops and Groovy file handling. Nextflow does all that for you. Am I missing something?

[LindoNkambule]

Apologies if my explanation of what I'd like to do is not clear. For each chromosome, there are 3 files: CSV, INPUT, REFERENCE. What I would like to do is to run basicExample on scattered mode (splitting CSV by line), but INPUT and REFERENCE do not change. Here is my attempt (which does not work):

#!/usr/bin/env nextflow

nextflow.enable.dsl=2

process basicExample {
    debug true
    tag "processing common ${irg}-${org}"

    input:
    tuple val(irg), val(org), val(input), val(ref)

    script:
    """
    echo --lower-bound ${irg} --upper-bound ${org} --input ${input} --ref ${ref}
    """
}


params.output_path = "/path/to/my/files"

workflow {
    Channel.from(1..23) \
        | map { i ->
            chunk_regions = Channel.fromPath("${params.output_path}/chunks/chunks_chr${i}.txt")
                                    .splitCsv(sep:'\t', header:['index', 'chrom', 'irg', 'org'])
                                    // .map{row -> [ row.irg, row.org ]}
                                    .map{row -> [ row[0], row[1] ]}
            input_vcf = "${params.output_path}/reference/input_chr${i}.bcf"
            reference_vcf = "${params.output_path}/inputs/ref_chr${i}.bcf"
            [ chunk_regions[0], chunk_regions[1], input_vcf, reference_vcf ]
        } \
        | basicExample
    

[mribeirodantas]

Isn't the following what you want?

params.ref = 'ref_file'
params.input = 'input.txt'

process FOO {
  debug true

  input:
  tuple val(name), val(value)
  path ref
  path input

  output:
  stdout

  script:
  """
  echo $name $value with $ref and $input
  """
}

workflow {
  Channel
    .fromPath('file.csv')
    .splitCsv(header: true)
    .set { csv_ch }

  FOO(csv_ch, file(params.ref), file(params.input))
}

If you have multiple files, you can always use globbing to get all of them and map+groupTuple to group them by index and do what I did above.