splitCsv map with text, join with another channel, nextflow

Question

I have a csv file which describes number of chunks per chromosome, like;

chrom,chunks
chr1,"CHUNKS=([1]=""0e6 3e6"" [2]=....)"
chr2,"CHUNKS=([1]=""0e6 3e6"" [2]=....)"
chr3,"CHUNKS=([1]=""0e6 3e6"" [2]=....)"
.....

I could crate a channel as follows,

channel
    .fromPath('chunks3.csv')
    .splitCsv(header:true)
    .map{row -> tuple(row.chrom,row.chunks)}
    .set{chunk_ch}

It gives me chromosome with respective chunk definition.

But when I try to join this channel with another channel, it does not work.

For example I have a channel "phased". It is a output channel created from a process.

channel
    .fromFilePairs('/some/path/genetic_map_chr*_combined_b37.txt',size:1)
    .map {group_key, file_list -> tuple(group_key.replaceFirst(/^*genetic_map_/,""),file_list.first())}
    .set {genetic_map}

channel
    .fromFilePairs('/some/path/*hrc.r1-1.ega.grch37.chr*.{hap.gz,legend.gz,samples}',size:3)
    .map {group_key,file_list -> tuple(group_key.replaceAll(/^.*.hrc\.r1-1\.ega\.grch37\.|\.haplotypes\.noibd|\.haplotypes/,""),file_list)}
    .set {ref_panel}

qc1_chr
    .map{chrom,bfiles -> tuple(chrom,bfiles.findAll{it.baseName.endsWith('step10')})}
    .join(ref_panel).join(genetic_map)
    .into {shape_in;imp_in}

process phasing {
    publishDir params.out, mode:'copy'
    
    input:
    tuple val(chrom), path(input_bed), path(input_ref), path(input_map) from shape_in
    
    output:
    tuple val("${chrom}"),path("${chrom}.*") into phased

    script:
    def (bed, bim, fam) = input_bed
    def (haplotype, legend, sample) = input_ref
    """
    shapeit --thread "${task.cpus}" --input-bed "${bed}" "${bim}" "${fam}" --input-map "${input_map}" --input-ref "${haplotype}" "${legend}" "${sample}" --duohmm --
    output-max "${chrom}.step10" --seed 54321
    """
}

Which emits something like:

phased.view()
[chr1,[file1]
chr2,[file2]
chr3,[file3]....]

If I try chunk_ch.join(phased) it gives nothing.

Any help?

Answer 1

Below is a reproducible example. I tried to simulate the same channels using chr1 as an example.

#!/usr/bin/env nextflow

// Simulate the chunk channel
channel
    .fromPath('chunk.csv')
    .splitCsv(header:true)
    .map{row -> tuple(row.chrom,row.chunks)}
    .set{chunk_ch}

// Simulate chr and chr_path as a single entry list
channel.from(["chr1", "chr1.txt"]).toList().map{tuple(it[0], it[1])}.set{shape_in}

// uncomment to view the content
// shape_in.view()

// process of phased_ch. I modified it to output the chr and chr_path as given in your example
process phased_ch {
  echo true
  input:
  tuple val(chr), val(chr_path) from shape_in
  output:
  tuple val("${chr}"), path("${chr_path}") into phased
  script:
  """
  echo ${chr}
  touch ${chr_path}
  """
}

// joining the two channels to produce the correct join
chunk_ch.join(phased).view()

// will produce 
// [chr1, "CHUNKS=([1]=""0e6 3e6"" [2]=....)", /media/Documents/Nextflow/work/8e/2fcd9a00b04f0f5d44711068d68a0f/chr1.txt]

chunk.csv contains

chrom,chunks
chr1,"CHUNKS=([1]=""0e6 3e6"" [2]=....)"
chr2,"CHUNKS=([1]=""0e6 3e6"" [2]=....)"
chr3,"CHUNKS=([1]=""0e6 3e6"" [2]=....)"

In your case, you should verify what are the contents of chunk_ch and phased on output. The first key of join() should be similar otherwise it won't give anything. From the documentation of join:

The join operator creates a channel that joins together the items emitted by two channels for which exists a matching key. The key is defined, by default, as the first element in each item emitted.