text: user16449: This is my first nextflow pipeline and i am struggling with the .fromFilePairs concept of Nextflow.
  I want to process paired-end read files and align them against a reference genome. I use Hisat2 so the index files come in form of 8 files with the suffix .ht2. Hisat needs the file path and the filename without ".ht2". All worked well with a single-end version of the process. The reference also has to be passed through a channel because i use a hisat docker-container and otherwise the files are not mounted.
  The idea is to pass read-pairs to a preprocessing step and then pass them (as tuples, like a Channel.fromPairs()) into the hisat process and then further downstream.
  My problem is all file-pairs are processed during the first step but only the first pair is passed further downstream. All downstream steps finish with the one pair and the rest simply stays in the output-folder of the preprocessing step.
  I tried to simply skip the first step to see if the "connection" between the first and second process is not working but the Channel which works for the first step does not for the hisat step. So i think there is a problem with the combination of multiple read-pairs as input but only one reference file?!
  process hisat2_paired {
    tag "$sample_id"
   
    container "${params.container_hisat}"
    publishDir "${params.out_dir}/Nextflow_output/hisat/"
  
    input:
    tuple val(sample_id), path(input_files)
    path( index_files )
  
    output:
    path('*.sam') , emit: hisat2_sam_out_ch
    path('*.txt')
  
    script:
    """
    INDEX=`find -L ${index_files}/ -name "*.1.ht2" | sed 's/.1.ht2//'` 
    hisat2 --rna-strandness RF \
            -x \\\$INDEX \
            -1 ${input_files[0]} \
        -2 ${input_files[1]} \
            -S ${sample_id}.sam &> ${sample_id}.hisat_summary.txt
    """
  
  
  workflow {
  
  Channel
      .fromFilePairs( "${params.reads_folder}/*_{R1,R2}*f*q.gz")
      .ifEmpty { exit 1, "no fastq files found at given path" }
      .view()
      .set{ read_pair_ch }
  
  Channel
      .fromPath( params.hisat2_index )
      .ifEmpty { exit 1, "no hisat2 index files found - path was empty" }
      //.view()
      .set { hisat2_index_ch }
  
  hisat2_paired( read_pair_ch, hisat2_index_ch )
  
  }
  
  
  I started the workflow with 3 test file-pairs and the Output on the CLI states that only one process is started although if i look at the Channel.view() all 3 file pairs are printed.
  executor >  local (1)
  [2f/30a4d6] process > hisat2_paired (MC3T3-Ko1_S2_L001) [  0%] 0 of 1
  [MC3T3-Ko1_S2_L001, [/mnt/c/Path/to/MC3T3-Ko1_S2_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko1_S2_L001_R2_001.fastq.gz]]
  [MC3T3-Ko2_S4_L001, [/mnt/c/Path/to/MC3T3-Ko2_S4_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko2_S4_L001_R2_001.fastq.gz]]
  [MC3T3-Ko3_S6_L001, [/mnt/c/Path/to/MC3T3-Ko3_S6_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko3_S6_L001_R2_001.fastq.gz]]
  
  I was already thinking about memory and resources. Since I prototype on my local PC i set the queue size in the config file to 2. I know that my hisat process needs about 7Gb RAM (for Mus.musculus) per process so there should be enough resources. Also the problem was that the local executor is not even trying to start a second process.
  
  user10775: 
  So i think there is a problem with the combination of multiple
  read-pairs as input but only one reference file?!
  
  Correct. The problem is that the hisat2_index_ch is a queue channel, but what you want is a value channel so that the (file) values can be read an unlimited number of times without being consumed. Assuming params.hisat2_index is intended to supply a glob pattern, you could simply call the collect operator to produce a list of files as the sole emission:
  Channel
      .fromPath( params.hisat2_index )
      .collect()
      .set { hisat2_index_ch }
  
  You would then need to update your process definition accordingly. Alternatively, have your workflow create the HISAT2 index files for you. This simplifies things considerably, by requiring only a single reference FASTA file as input. Note that value channels are implicitly created as output for a process whose inputs are all value channels. This means in the example below, hisat2_index.out is also a value channel:
  params.publish_dir = './results'
  
  params.container_hisat = 'quay.io/biocontainers/hisat2:2.2.1--h87f3376_4'
  
  params.reads = '*_R{1,2}.fastq.gz'
  params.ref_fasta = 'chr1.fasta'
  
  
  process hisat2_index {
  
      container params.container_hisat
      cpus 8
  
      input:
      path fasta
  
      output:
      tuple val(fasta.baseName), path("${fasta.baseName}.*.ht2")
  
      """
      hisat2-build \\
          -p ${task.cpus} \\
          "${fasta}" \\
          "${fasta.baseName}"
      """
  }
  
  process hisat2_paired {
  
      tag { sample_id }
  
      container params.container_hisat
      publishDir "${params.publish_dir}/hisat"
  
      input:
      tuple val(sample_id), path(input_files)
      tuple val(idx), path('hisat2-idx/*')
  
      output:
      path('*.sam'), emit: hisat2_sam
      path('*.txt')
  
      script:
      def (fq1, fq2) = input_files
  
      """
      hisat2 \\
        --rna-strandness RF \\
        -x "hisat2-idx/${idx}" \\
        -1 ${fq1} \\
        -2 ${fq2} \\
        -S ${sample_id}.sam \\
        &> ${sample_id}.hisat_summary.txt
      """
  }
  
  workflow {
  
      read_pairs = Channel.fromFilePairs( params.reads )
      ref_fasta = file( params.ref_fasta )
  
      hisat2_index( ref_fasta )
  
      hisat2_paired( read_pairs, hisat2_index.out )
  }
  
  
original_text: 16449: <p>This is my first nextflow pipeline and i am struggling with the .fromFilePairs concept of Nextflow.</p>
  <p>I want to process paired-end read files and align them against a reference genome. I use Hisat2 so the index files come in form of 8 files with the suffix .ht2. Hisat needs the file path and the filename without &quot;.ht2&quot;. All worked well with a single-end version of the process. The reference also has to be passed through a channel because i use a hisat docker-container and otherwise the files are not mounted.</p>
  <p>The idea is to pass read-pairs to a preprocessing step and then pass them (as tuples, like a Channel.fromPairs()) into the hisat process and then further downstream.</p>
  <p>My problem is all file-pairs are processed during the first step but only the first pair is passed further downstream. All downstream steps finish with the one pair and the rest simply stays in the output-folder of the preprocessing step.</p>
  <p>I tried to simply skip the first step to see if the &quot;connection&quot; between the first and second process is not working but the Channel which works for the first step does not for the hisat step. So i think there is a problem with the combination of multiple read-pairs as input but only one reference file?!</p>
  <pre><code>process hisat2_paired {
    tag &quot;$sample_id&quot;
   
    container &quot;<span class="math-container">${params.container_hisat}"
    publishDir "$</span>{params.out_dir}/Nextflow_output/hisat/&quot;
  
    input:
    tuple val(sample_id), path(input_files)
    path( index_files )
  
    output:
    path('*.sam') , emit: hisat2_sam_out_ch
    path('*.txt')
  
    script:
    &quot;&quot;&quot;
    INDEX=`find -L ${index_files}/ -name &quot;*.1.ht2&quot; | sed 's/.1.ht2//'` 
    hisat2 --rna-strandness RF \
            -x \\\$INDEX \
            -1 <span class="math-container">${input_files[0]} \
        -2 $</span>{input_files[1]} \
            -S <span class="math-container">${sample_id}.sam &amp;&gt; $</span>{sample_id}.hisat_summary.txt
    &quot;&quot;&quot;
  
  </code></pre>
  <pre><code>workflow {
  
  Channel
      .fromFilePairs( &quot;${params.reads_folder}/*_{R1,R2}*f*q.gz&quot;)
      .ifEmpty { exit 1, &quot;no fastq files found at given path&quot; }
      .view()
      .set{ read_pair_ch }
  
  Channel
      .fromPath( params.hisat2_index )
      .ifEmpty { exit 1, &quot;no hisat2 index files found - path was empty&quot; }
      //.view()
      .set { hisat2_index_ch }
  
  hisat2_paired( read_pair_ch, hisat2_index_ch )
  
  }
  
  </code></pre>
  <p>I started the workflow with 3 test file-pairs and the Output on the CLI states that only one process is started although if i look at the Channel.view() all 3 file pairs are printed.</p>
  <pre><code>executor &gt;  local (1)
  [2f/30a4d6] process &gt; hisat2_paired (MC3T3-Ko1_S2_L001) [  0%] 0 of 1
  [MC3T3-Ko1_S2_L001, [/mnt/c/Path/to/MC3T3-Ko1_S2_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko1_S2_L001_R2_001.fastq.gz]]
  [MC3T3-Ko2_S4_L001, [/mnt/c/Path/to/MC3T3-Ko2_S4_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko2_S4_L001_R2_001.fastq.gz]]
  [MC3T3-Ko3_S6_L001, [/mnt/c/Path/to/MC3T3-Ko3_S6_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko3_S6_L001_R2_001.fastq.gz]]
  </code></pre>
  <p>I was already thinking about memory and resources. Since I prototype on my local PC i set the queue size in the config file to 2. I know that my hisat process needs about 7Gb RAM (for Mus.musculus) per process so there should be enough resources. Also the problem was that the local executor is not even trying to start a second process.</p>
   10775: <blockquote>
  <p>So i think there is a problem with the combination of multiple
  read-pairs as input but only one reference file?!</p>
  </blockquote>
  <p>Correct. The problem is that the <em>hisat2_index_ch</em> is a <a href="https://www.nextflow.io/docs/latest/channel.html#queue-channel" rel="nofollow noreferrer">queue channel</a>, but what you want is a <a href="https://www.nextflow.io/docs/latest/channel.html#value-channel" rel="nofollow noreferrer">value channel</a> so that the (file) values can be read an unlimited number of times without being consumed. Assuming <em>params.hisat2_index</em> is intended to supply a glob pattern, you could simply call the <a href="https://www.nextflow.io/docs/latest/operator.html#collect" rel="nofollow noreferrer">collect</a> operator to produce a list of files as the sole emission:</p>
  <pre><code>Channel
      .fromPath( params.hisat2_index )
      .collect()
      .set { hisat2_index_ch }
  </code></pre>
  <p>You would then need to update your process definition accordingly. Alternatively, have your workflow create the HISAT2 index files for you. This simplifies things considerably, by requiring only a single reference FASTA file as input. Note that value channels are implicitly created as output for a process whose inputs are all value channels. This means in the example below, <em>hisat2_index.out</em> is also a value channel:</p>
  <pre><code>params.publish_dir = './results'
  
  params.container_hisat = 'quay.io/biocontainers/hisat2:2.2.1--h87f3376_4'
  
  params.reads = '*_R{1,2}.fastq.gz'
  params.ref_fasta = 'chr1.fasta'
  
  </code></pre>
  <pre><code>process hisat2_index {
  
      container params.container_hisat
      cpus 8
  
      input:
      path fasta
  
      output:
      tuple val(fasta.baseName), path(&quot;${fasta.baseName}.*.ht2&quot;)
  
      &quot;&quot;&quot;
      hisat2-build \\
          -p <span class="math-container">${task.cpus} \\
          "$</span>{fasta}&quot; \\
          &quot;${fasta.baseName}&quot;
      &quot;&quot;&quot;
  }
  </code></pre>
  <pre><code>process hisat2_paired {
  
      tag { sample_id }
  
      container params.container_hisat
      publishDir &quot;${params.publish_dir}/hisat&quot;
  
      input:
      tuple val(sample_id), path(input_files)
      tuple val(idx), path('hisat2-idx/*')
  
      output:
      path('*.sam'), emit: hisat2_sam
      path('*.txt')
  
      script:
      def (fq1, fq2) = input_files
  
      &quot;&quot;&quot;
      hisat2 \\
        --rna-strandness RF \\
        -x &quot;hisat2-idx/<span class="math-container">${idx}" \\
        -1 $</span>{fq1} \\
        -2 <span class="math-container">${fq2} \\
        -S $</span>{sample_id}.sam \\
        &amp;&gt; ${sample_id}.hisat_summary.txt
      &quot;&quot;&quot;
  }
  </code></pre>
  <pre><code>workflow {
  
      read_pairs = Channel.fromFilePairs( params.reads )
      ref_fasta = file( params.ref_fasta )
  
      hisat2_index( ref_fasta )
  
      hisat2_paired( read_pairs, hisat2_index.out )
  }
  </code></pre>
  