Answers.Body: <h1>Update 3</h1>
  <p>While the code in <strong>Update 2</strong> works, @Steve pointed out that my approach may not be ideal in cloud environments (see comments). He also provided a <code>Nextflow</code> solution that works. I'm still not a fan of how <code>Nextflow</code> <code>output:</code> <code>Channels</code> work, but that may be from my inexperience.</p>
  <h1>Update 2</h1>
  <p>Pretty sure I got it this time. <strong>The following code works</strong>. I'm getting all of the <code>tuples</code> I was expecting.</p>
  <p>Here's the updated code for my <code>workflow</code> to create the tuples from each line in the <code>sample_tuple_file</code> I created from a <code>bash</code> <code>for</code> loop. I spent quite a while trying to get a combination of <code>.splitText()</code> and <code>.splitCSV()</code> to work, but for some reason, you cannot chain as follows: <code>it.splitText().splitCSV()</code>.</p>
  <pre><code>     GENERATE_FASTQS_PROC.out.sample_tuple_file
           .map {
              it.splitText()
           }
           .flatten()
           .map {
               toks = it.trim().split(',')
               sample_name = toks[0]
               fq = toks[1]
               rg = toks[2]
               tuple(sample_name, fq, rg)
               }
           .view()
          .set { sample_fq_tuples_ch }
  </code></pre>
  <hr />
  <h1>Update 1</h1>
  <p>I thought I had it, but I'm actually only getting ONE of the tuples from my <code>GENERATE_FASTQS_PROC.out.sample_tuple_file.map { }</code> statement (the last one). And I have no idea why...</p>
  <hr />
  <h1>Original posted answer</h1>
  <p>Despite @Steve's exceptional help and patience with me, I have really battled to figure out how to work with the <code>Nextflow</code> <code>output:</code> <code>Channels</code>, but simply could not figure out any solution using them.</p>
  <p>I finally figured out a super simple method that is much easier than dealing with <code>Nextflow</code> <code>Channels</code>, which I share here in case it will help others. I personally will probably use this method from now on, unless someone can convince me that's a bad idea. In my view, using my approach gives me complete control over the way outputs and inputs are handled, and I don't have to worry whether <code>Nextflow</code> is doing what I hope it's doing.</p>
  <p><strong>generate_fastqs.sh</strong>
  <code>generate_fastqs.sh</code> takes a single <code>.bam</code> -&gt; sorts by name -&gt; converts to .fastq (now w/ <code>samtools fastq</code>, thanks to @Steve), and splits them with <code>Unix</code> <code>split</code>. Then, at the end of the script, I create a single file with all of the <code>tuples</code> that I want to pass to my next <code>process</code>.</p>
  <p>Here are the key lines for these steps (including exit codes to ensure <code>Nextflow</code> receives error notifications). The last step is the key (for me, anyway):</p>
  <pre><code>...
  
  #####################
  # Sort .bam by name #
  #####################
  if ! time samtools sort -@ <span class="math-container">$n_threads -n -m $</span>{mem_per_thread}G <span class="math-container">$sample_input &gt; $</span>sample_sorted_by_name; then
      echo &quot;ERROR: sort failed&quot;
      exit 1
  fi
  
  ...
  
  #############################
  # Create interleaved .fastq #
  #############################
  if ! time samtools fastq -o <span class="math-container">$fq -s /dev/null -0 /dev/null --reference $</span>original_ref $sample_sorted_by_name; then
      echo &quot;ERROR: bamtofastq failed&quot;
      exit 1
  fi
  
  ...
  
  ################
  # Split fastqs #
  ################
  if ! time split --additional-suffix=&quot;.fastq&quot; -dl <span class="math-container">$n_lines $</span>fq &quot;<span class="math-container">${split_dir}/$</span>{sample_name}.interleaved_R1_R2.split.&quot;; then
      echo &quot;ERROR: split failed&quot;
      exit 1
  fi
  
  ...
  
  ###########################################################
  # Store sample name, .fastq, and RG to file for alignment #
  ###########################################################
  RG_file=&quot;${sample_name}.tuples.txt&quot;
  
  # Create empty file
  &gt; <span class="math-container">$RG_file
  for fq_file in $</span>PWD/<span class="math-container">${split_dir}/*
  do
  echo "$</span>{sample_name},<span class="math-container">$fq_file,$</span>RG&quot; &gt;&gt; $RG_file    # &lt;-- Creating future tuples in file, here (comma separated).
  done
  </code></pre>
  <p><strong>GENERATE_FASTQS_PROC</strong></p>
  <p>In <code>GENERATE_FASTQS_PROC</code>, <code>emit</code> one simple file--the tuple file I created myself with a simple <code>bash</code> loop (so easy and feels so good):</p>
  <pre><code>    emit:
          path(&quot;*.tuples.txt&quot;), emit: sample_tuple_file
  </code></pre>
  <p><strong>Workflow</strong></p>
  <p>In my <code>workflow</code>, I call <code>GENERATE_FASTQS_PROC</code>, access the file from the <code>Channel</code>, read each line, and create a <code>Nextflow</code> tuple for each line (again, the lines I so easily created in a <code>bash</code> <code>for</code> loop).</p>
  <pre><code>    /*
       * Generate fastq files from the original .(cr|b)am file and split into
       * mini .fastq files with `n_reads_per_run` in each mini .fastq
       */
      GENERATE_FASTQS_PROC(sample_input_file_ch, n_reads_per_run, original_ref)
  
  
      /*
       * Get sample name, fastq, and read group (RG) from the tuple file using Nextflow
       */
  
       GENERATE_FASTQS_PROC.out.sample_tuple_file
          .map{
              sample_tuple_file -&gt;
                  sample_tuple_file.withReader {          // Open file
                      it.eachLine {                       // Loop over each line in file
                          line -&gt;
                              toks = line.split(',')      // Split by comma
                              sample_name = toks[0]       // Get each item
                              fq = toks[1]
                              rg = toks[2]
  
                              tuple(sample_name, fq, rg)  // Create the simple tuple that has eluded me for two days!
                      }
                  }
          }
          .set { sample_fq_tuples_ch }                    // Assign it to a new Channel name and move along!
  
  
      /*
       * Align fastq file pairs
       */
  
      ALIGN_FASTQ_BWA_PROC(sample_fq_tuples_ch, align_to_ref,
                               align_to_ref_tag, original_ref, output_format)
  <span class="math-container">```</span>
  </code></pre>
  
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-13T01:51:18.667Z
Answers.Text: Ugh. I thought I had it, but I confused myself with my own print statements. I'm actually only getting ONE of the tuples from my `GENERATE_FASTQS_PROC.out.sample_tuple_file.map { }` statement. And I have no idea why...
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-13T03:11:50.917Z
Answers.Text: Pretty sure I got it working with update #2. Hopefully others can learn from my mistakes and final victory.
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-13T14:52:23.093Z
Answers.Text: I can see you have put in quite a bit of effort here, but this is not something I would recommend to others. This is a solution that will work on a local filesystem _only_, where all files exist under a common root. This is what makes it possible to pick them up later on if you have a pointer to them (your tuples.txt contains file pointers). Since the required files are not declared in the output block, Nextflow will not attempt to "unstage" them (upload them to a remote object store, like a bucket) when you deploy your workflow to the cloud. They will just be lost.
Answers.UserDisplayName: Steve
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-13T15:02:10.017Z
Answers.Text: Thanks for including your `generate_fastqs.sh`. I have a better picture now of what you are trying to do. I will try to update my answer with a working solution soon. I just need to grab a couple of test BAMs. I should have a couple somewhere..
Answers.UserDisplayName: Steve
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-14T02:56:51.773Z
Answers.Text: True. I hadn’t considered cloud file systems. I was worried you’d burst my bubble. :-)
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-14T03:10:28.790Z
Answers.Text: Couldn’t I still emit the files you mention, but use my method to generate tuples, etc? Or, couldn’t I publish results and access them that way? I realize this breaks convention, but it appears that, while Nextflow has many advantages,  Nextflow’s output Channels are fundamentally convoluted, flawed, and maybe even dangerous. I’d love to be proven wrong. I don’t know Nextflow’s internals, but seems like this could be easily fixed. e.g., if every process instance assigned an ID to the process that we could filter by upon return, I could trust that NF is giving me what I think I’m getting.
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-15T06:33:42.097Z
Answers.Text: You could perhaps. But that would not be a pleasant experience at all and would produce a convoluted solution. I don't have any real problem with Nextflow Channels. I think they're great and they make writing complex workflows possible. I've updated my answer above. I think it's ok as an example but there are some caveats listed. You may or may not need to address these. But it will likely need some tinkering before you can run with it.
Answers.UserDisplayName: Steve
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-13T01:24:35.137Z
Answers.LastActivityDate: 2022-02-16T15:02:17.650Z
Answers.LastEditDate: 2022-02-16T15:02:17.650Z
Answers.UserDisplayName: Mark Ebbert
Answers.Body: <p>Nextflow channels guarantee that items are delivered in the same order as they are sent. So if a process declares two (or more) output channels, the items that are emitted should already be synchronized. This is true unless of course one of the channels sets the <a href="https://www.nextflow.io/docs/latest/process.html#optional-output" rel="nofollow noreferrer">optional output</a> attribute.</p>
  <blockquote>
  <ol>
  <li>How do I synchronize the emitted <code>@RG</code> file (<code>sample_RG_file</code>) with the
  split <code>.fastqs</code>?</li>
  </ol>
  </blockquote>
  <p>Since the <a href="https://www.nextflow.io/docs/latest/operator.html#splitfastq" rel="nofollow noreferrer">splitFastq</a> operator <em>can</em> create <em>n</em> outputs for each input, you need some way to <a href="https://www.nextflow.io/docs/latest/operator.html#combine" rel="nofollow noreferrer">combine</a> the @RG files using some key. This key, for example, could be the basename of the sample input file. This, I think, would work nicely given the input file should just have a '.bam' extension. Another option, would be to supply some other value in the input declaration and just use that in your output declaration. If we went with the former, your output declaration might look like:</p>
  <pre><code>output:
  tuple val(sample_input_file.baseName), path(&quot;sample_R1.fastq&quot;), path(&quot;sample_R2.fastq&quot;), \
      emit: fastqs
  tuple val(sample_input_file.baseName), path(&quot;sample_RG.txt&quot;), \
      emit: sample_RG
  </code></pre>
  <p>Then, the body of your workflow might look like:</p>
  <pre><code>workflow {
  
      reads = GENERATE_FASTQS_PROC(sample_input_file_ch, original_ref)
          
      reads.fastqs
          .splitFastq( by: params.reads_per_job, pe: true, file: true )
          .combine( reads.sample_RG, by: 0 )
          .map { sample, fq1, fq2, rg_txt -&gt; tuple(fq1, fq2, rg_txt) }
          .set { fq_split_reads_and_RG_ch }
  
      ...
  }
  </code></pre>
  <p>You'll notice that I removed your fuzzy glob patterns from your output declaration above. Your workflow definition suggests that the <code>generate_fastqs.sh</code> script produces a single pair of FASTQ files and a single readgroup file for a given set of inputs. However, as you know, glob patterns can match one or more output files. When declaring output files, it's always best to name these guys explicitly. If you do use a glob pattern and for some reason your process was to produce two files that match a given pattern, you'll instead get back a tuple, which would break the pipeline.</p>
  <blockquote>
  <ol start="2">
  <li>Is there a better way?</li>
  </ol>
  </blockquote>
  <p>Since you need the @RG tags with the split FASTQ files, I don't think there is. However, I generally avoid using the <a href="https://www.nextflow.io/docs/latest/operator.html#splitting-operators" rel="nofollow noreferrer">splitting</a> operators on large files. My preference is always to pass &quot;work&quot; off (including IO) to a separate process. A separate process can also provide increased flexibility if/when requirements change.</p>
  <blockquote>
  <ol start="3">
  <li>Does Nextflow not stage files until the actual .bash script starts? I tried to operate
  on the sample_RG_file within the script using Groovy, but got an error
  that the file didn't exist.</li>
  </ol>
  </blockquote>
  <p>Nextflow evaluates the code in the script definition before returning the script string to be executed. This is why it's possible to implement a <a href="https://www.nextflow.io/docs/latest/process.html#conditional-scripts" rel="nofollow noreferrer">conditional script</a>. Basically, you want to avoid all IO operations here. The script string to be executed has not even been decided at this point.</p>
  <hr />
  <p><strong>Update</strong></p>
  <p>I'm not sure how you plan to operate per sample. IIUC, <code>[&lt;all_fastqs_for_sample_A&gt;]</code> is basically a Channel of lists, where each list contains all split FASTQ files for a given sample. Using <a href="https://www.nextflow.io/docs/latest/operator.html#map" rel="nofollow noreferrer">map</a> here to pull out the split ID numbers followed by the <a href="https://www.nextflow.io/docs/latest/operator.html#grouptuple" rel="nofollow noreferrer">groupTuple</a> operator might not give the expected results. You might also run into the problems with calling <a href="https://www.nextflow.io/docs/latest/operator.html#grouptuple" rel="nofollow noreferrer">groupTuple</a>, since, without a <code>size</code> attribute, it will basically wait for all outputs before emitting the grouped result:</p>
  <blockquote>
  <p>You should always specify the number of expected elements in each
  tuple using the <code>size</code> attribute to allow the <code>groupTuple</code> operator to
  stream the collected values as soon as possible. However there are use
  cases in which each tuple has a different size depending on the
  grouping key. In this case use the built-in function <code>groupKey</code> that
  allows you to create a special grouping key object to which it’s
  possible to associate the group size for a given key.</p>
  </blockquote>
  <p>Part of the problem is that we don't know what the <code>generate_fastqs.sh</code> script does and how the other scripts work. If you are aligning using BWA MEM and have <a href="http://www.htslib.org/doc/samtools.html" rel="nofollow noreferrer">samtools</a> available in your $PATH, it might even be easier to produce interleaved output (since BWA MEM can read interleaved FASTQs) using the <a href="http://www.htslib.org/doc/samtools-fasta.html" rel="nofollow noreferrer">samtools fastq</a> command. This would save some work having to group each list of FASTQs into pairs, ensuring they are ordered as expected. This would be reasonably involved.</p>
  <p>Not sure what BAMs you need to process, or if you're looking for a more generic solution, but I thought I'd throw in an alternative solution you might find works well for you or that you may not have yet considered. Since BAMs often contain multiple readgroups, I find that if I split by readgroup (using <a href="http://www.htslib.org/doc/samtools-split.html" rel="nofollow noreferrer">samtools split</a>, <a href="https://www.nextflow.io/docs/latest/operator.html#flatten" rel="nofollow noreferrer">flatten</a> the output Channel, then <a href="http://www.htslib.org/doc/samtools-collate.html" rel="nofollow noreferrer">samtools collate</a> followed by <a href="http://www.htslib.org/doc/samtools-fasta.html" rel="nofollow noreferrer">samtools fastq</a>, I can achieve somewhat uniform mapping walltimes (assuming the input BAMs are similar) without needing to chunk the reads and then merge the alignments back together. If your input BAMs are not similar at all, then you will likely need to chunk to generate uniform alignment jobs.</p>
  <p>Below is an example, that shows how to create a groupKey, followed by the transpose, join/combine and groupTuple pattern. There are some caveats and there is some room for optimization:</p>
  <pre><code>nextflow.enable.dsl=2
  
  params.original_ref = './ref.fasta'
  params.reads_per_job = 10000
  
  
  process samtools_view_header {
  
      tag { &quot;<span class="math-container">${sample_name}:$</span>{bam.name}&quot; }
  
      cpus 1
      memory 1.GB
  
      input:
      tuple val(sample_name), path(bam)
  
      output:
      tuple val(sample_name), path(&quot;${bam.baseName}.header.txt&quot;)
  
      &quot;&quot;&quot;
      samtools view \\
          -H \\
          -o &quot;<span class="math-container">${bam.baseName}.header.txt" \\
          "$</span>{bam}&quot;
      &quot;&quot;&quot;
  }
  
  process samtools_name_sort {
  
      tag { &quot;<span class="math-container">${sample_name}:$</span>{bam.name}&quot; }
  
      cpus 4
      memory 8.GB
  
      input:
      tuple val(sample_name), path(bam)
  
      output:
      tuple val(sample_name), path(&quot;${bam.baseName}.nsorted.bam&quot;)
  
      script:
      def avail_mem = task.memory ? task.memory.toGiga().intdiv(task.cpus) : 0
      def mem_per_thread = avail_mem ? &quot;-m ${avail_mem}G&quot; : ''
  
      &quot;&quot;&quot;
      samtools sort \\
          -@ &quot;<span class="math-container">${task.cpus - 1}" \\
          $</span>{mem_per_thread} \\
          -n \\
          -o &quot;<span class="math-container">${bam.baseName}.nsorted.bam" \\
          -T "$</span>{bam.baseName}.nsorted&quot; \\
          &quot;${bam}&quot;
      &quot;&quot;&quot;
  }
  
  process samtools_fastq {
  
      tag { &quot;<span class="math-container">${sample_name}:$</span>{bam.name}&quot; }
  
      cpus 1
      memory 2.GB
  
      input:
      tuple val(sample_name), path(bam)
  
      output:
      tuple val(sample_name), path(&quot;${bam.baseName}.fastq&quot;)
  
      &quot;&quot;&quot;
      samtools fastq \\
          -O \\
          -T RG,BC \\
          -0 /dev/null \\
          &quot;<span class="math-container">${bam}" \\
          &gt; \\
          "$</span>{bam.baseName}.fastq&quot;
      &quot;&quot;&quot;
  }
  
  process split_fastq {
  
      tag { &quot;<span class="math-container">${sample_name}:$</span>{fastq.name}&quot; }
  
      cpus 1
      memory 1.GB
  
      input:
      tuple val(sample_name), path(fastq)
  
      output:
      tuple val(sample_name), path(&quot;<span class="math-container">${fastq.baseName}.$</span>{/[0-9]/*5}.fastq&quot;)
  
      &quot;&quot;&quot;
      split \\
          --suffix-length=5 \\
          --additional-suffix=&quot;.fastq&quot; \\
          -d \\
          -l &quot;<span class="math-container">${params.reads_per_job*4}" \\
          "$</span>{fastq}&quot; \\
          &quot;${fastq.baseName}.&quot;
      &quot;&quot;&quot;
  }
  
  process bwa_index {
  
      tag { fasta.name }
  
      cpus 1
      memory 12.GB
  
      input:
      path fasta
  
      output:
      tuple val(fasta.name), path(&quot;${fasta}.{amb,ann,bwt,pac,sa}&quot;)
  
      &quot;&quot;&quot;
      bwa index &quot;${fasta}&quot;
      &quot;&quot;&quot;
  }
  
  process bwa_mem {
  
      tag { &quot;<span class="math-container">${sample_name}:$</span>{fastq.name}&quot; }
  
      cpus 8
      memory 12.GB
  
      input:
      tuple val(sample_name), path(fastq), path(header)
      tuple val(idxbase), path(bwa_index)
  
      output:
      tuple val(sample_name), path(&quot;${fastq.baseName}.bam&quot;)
  
      script:
      def task_cpus = task.cpus &gt; 1 ? task.cpus - 1 : task.cpus
  
      &quot;&quot;&quot;
      bwa mem \\
          -p \\
          -t <span class="math-container">${task_cpus} \\
          -C \\
          -H &lt;(grep "^@RG" "$</span>{header}&quot;) \\
          &quot;<span class="math-container">${idxbase}" \\
          "$</span>{fastq}&quot; |
      samtools view \\
          -1 \\
          -o &quot;${fastq.baseName}.bam&quot; \\
          -
      &quot;&quot;&quot;
  }
  
  process samtools_coordinate_sort {
  
      tag { &quot;<span class="math-container">${sample_name}:$</span>{bam.name}&quot; }
  
      cpus 4
      memory 8.GB
  
      input:
      tuple val(sample_name), path(bam)
  
      output:
      tuple val(sample_name), path(&quot;${bam.baseName}.csorted.bam&quot;)
  
      script:
      def avail_mem = task.memory ? task.memory.toGiga().intdiv(task.cpus) : 0
      def mem_per_thread = avail_mem ? &quot;-m ${avail_mem}G&quot; : ''
  
      &quot;&quot;&quot;
      samtools sort \\
          -@ &quot;<span class="math-container">${task.cpus - 1}" \\
          $</span>{mem_per_thread} \\
          -o &quot;<span class="math-container">${bam.baseName}.csorted.bam" \\
          -T "$</span>{bam.baseName}.csorted&quot; \\
          &quot;${bam}&quot;
      &quot;&quot;&quot;
  }
  
  process samtools_merge {
  
      tag { sample_name }
  
      cpus 1
      memory 4.GB
  
      input:
      tuple val(sample_name), path(bams)
  
      output:
      tuple val(sample_name), path(&quot;${sample_name}.bam{,.bai}&quot;)
  
      &quot;&quot;&quot;
      samtools merge \\
          -c \\
          -p \\
          &quot;<span class="math-container">${sample_name}.bam" \\
          $</span>{bams}
      samtools index \\
          &quot;${sample_name}.bam&quot;
      &quot;&quot;&quot;
  }
  
  workflow {
  
      original_ref = file( params.original_ref )
      bwa_index( original_ref )
  
      Channel.fromPath('./data/*.bam')
          | map { tuple( it.baseName, it ) }
          | set { sample_bam_files }
  
      samtools_view_header( sample_bam_files )
  
      sample_bam_files
          | samtools_name_sort
          | samtools_fastq
          | split_fastq
          | map { sample_name, fastq_files -&gt;
              tuple( groupKey(sample_name, fastq_files.size()), fastq_files )
          }
          | transpose()
          | combine( samtools_view_header.out, by: 0 )
          | set { realignment_inputs }
  
      bwa_mem( realignment_inputs, bwa_index.out )
          | samtools_coordinate_sort
          | groupTuple()
          | map { sample_name, bam_files -&gt;
              tuple( sample_name.toString(), bam_files )
          }
          | samtools_merge
          | view()
  }
  </code></pre>
  <p>There are some caveats however, in no particular order:</p>
  <ul>
  <li><p>It unravels your shell scripts. I think this is a better solution though because it gives you control over each of the components.</p>
  </li>
  <li><p>No CRAM support. But this should be easy to add if needed.</p>
  </li>
  <li><p>A single process to read a BAM header is wasteful, especially if the input BAM is large. More time will be spent localizing the file than actually doing work.</p>
  </li>
  <li><p>Replace <code>samtools sort -n</code> with <code>samtools collate</code>. There's really no need to name sort. You may prefer to pipe the output of <code>samtools collate</code> into <code>samtools fastq</code> for avoid writing an extra file. This might be useful if you need to operate on an extremely large number of large BAMs.</p>
  </li>
  <li><p>The 'samtools_fastq' step should write gzipped output with low compression. This means replacing coreutils <code>split</code> with something else. A solution using Biopython could be useful here to read and write gzipped output in a single pass.</p>
  </li>
  <li><p>The last 'map' command in the workflow is unnecessary. I just thought I'd show that calling <code>.toString()</code> on the GroupKey gives you back an actual String object.</p>
  </li>
  <li><p>Also, it's not necessary to pass around a tuple of 'sample name' and paths as process inputs and outputs. There's no harm in doing this though. When you are more confident with Nextflow Channels, refactoring these processes should be easy.</p>
  </li>
  <li><p>My <code>samtools sort</code> processes ensure that they do not oversubscribe the nodes that they land on. They use approximately 320% - 380% CPU, depending on the size of the BAM. The resource requests for the other processes are not optimized at all.</p>
  </li>
  <li><p>With newer versions of samtools, BAM indexing can be done at the same time as merging, with the <code>--write-index</code> option.</p>
  </li>
  </ul>
  
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-12T02:46:19.437Z
Answers.Text: Thanks, @Steve. You've made several important points. Question about your suggested `workflow` code: wouldn't `.splitFastq` fail since `reads.fastqs` contains a tuple including the `sample_file_input.baseName`? I've been battling how to work with the `tuples` of `tuples` in various scenarios (e.g., in this situation and after a `.groupTuple()`).
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-12T02:47:57.633Z
Answers.Text: Also, per your suggestion, I've been working to have `GENERATE_FASTQS_PROC` split the `.fastqs` itself and emit those. That's causing me a headache. The level of abstraction in `Nextflow` has really made these details quite elusive for me.
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-12T03:43:26.473Z
Answers.Text: I added an update to my question explaining what I feel like I should be able to do and what I ultimately want to get.
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-12T21:31:41.087Z
Answers.Text: Thanks, @Steve. I think the interleaved option might be the simplest. Great suggestion. Still, I'm shocked by how difficult it is to get files out of a `process` without mixing them between samples. What kills me is, I know all files for a single sample are safely contained in a unique directory, but I'm forced to mix them up with files from other samples in the same `Channel`, so I'm left with convoluted methods to try to distinguish them from each other. If I `emit` the `path` to the directories and then use `Channel.fromPath()`, it would be sooooo easy. But that's not possible, AFAIK.
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-12T21:33:44.030Z
Answers.Text: `generate_fastqs.sh` takes a single `.bam` -> sorts by name -> converts to `.fastq` (currently w/ `bedtools`), and splits them with `Unix` `split`. The files are stored safe and sound, all tidy-like in a unique directory, where I have clear control over what's generated and mixed together (unlike `Nextflow` `Channels`). Forgive the frustration. :-)
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-13T13:37:01.680Z
Answers.Text: Sorry @MarkEbbert, I was meant to let you know yesterday that I had updated my answer but got pulled in another direction before I could finish the update. WRT your first question above, [splitFastq](https://www.nextflow.io/docs/latest/operator.html#splitfastq) will work fine with a tuple that includes a sample identifier. For example, it will work with the output of [fromFilePairs](https://www.nextflow.io/docs/latest/channel.html#fromfilepairs) with `flat: true`.
Answers.UserDisplayName: Steve
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-13T14:12:39.217Z
Answers.Text: @MarkEbbert I struggled with the same thing when I first started with NF. I found that outputting tuples of files, prefixed with some identifier (usually a sample name) helps considerably when working with multiple channels.
Answers.UserDisplayName: Steve
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-16T14:57:37.613Z
Answers.Text: This was a huge help, @Steve. You've also made several additional helpful suggestions. It had never dawned on me that `samtools` was oversubscribing because `-@` is for *additional* threads, though that should have been completely obvious. Really appreciate your help. I'm still not a fan of NF `output` `Channels`, but maybe I'll get there. :-)
Answers.UserDisplayName: Mark Ebbert
Answers.ContentLicense: CC BY-SA 4.0
Answers.CreationDate: 2022-02-11T16:54:55.763Z
Answers.LastActivityDate: 2022-02-15T06:34:44.010Z
Answers.LastEditDate: 2022-02-15T06:34:44.010Z
Answers.UserDisplayName: Steve
Body: <p>I have a workflow that needs to:</p>
  <ol>
  <li>Generate <code>.fastq</code> files from <code>.bams</code> (while preserving the <code>@RG</code> group from the original bam)</li>
  <li>Split the <code>.fastqs</code></li>
  <li>Align</li>
  <li>etc.</li>
  </ol>
  <p>The only way I could think to preserve the <code>@RG</code> group was to print it to a file during step 1 and <code>emit</code> the file. I can't figure out how to synchronize the emitted <code>@RG</code> file with the <code>.fastqs</code>, however. I've tried several variations without luck.</p>
  <p>Questions:</p>
  <ol>
  <li>How do I synchronize the emitted <code>@RG</code> file (<code>sample_RG_file</code>) with the split <code>.fastqs</code>?</li>
  <li>Is there a better way?</li>
  <li><strong>Bonus question</strong> (may be worth a separate post): does Nextflow not stage files until the actual .bash script starts? I tried to operate on the <code>sample_RG_file</code> within the <code>script</code> using <code>Groovy</code>, but got an error that the file didn't exist. (The code was working before I found I needed to synchronize <code>sample_RG_file</code> with the</li>
  </ol>
  <p>Other variations:</p>
  <ol>
  <li>The code was 'working' before I found I needed to synchronize <code>sample_RG_file</code> with the split <code>.fastqs</code>. I was originally passing <code>GENERATE_FASTQS_PROC.out.sample_RG</code> as a separate parameter to <code>ALIGN_FASTQ_BWA_PROC</code>, but it was always passing the same (first) file.</li>
  <li>I also tried <code>GENERATE_FASTQS_PROC.out.sample_RG.flatten()</code> in the <code>.map</code> operator, but that had the same error.</li>
  </ol>
  <p><strong>Current error:</strong></p>
  <p>The code below fails with the following error:</p>
  <blockquote>
  <p>Error executing process &gt; 'REALIGN_SAMPLES_WF:ALIGN_FASTQ_BWA_PROC (1)'</p>
  </blockquote>
  <blockquote>
  <p>Caused by:
  Not a valid path value type: groovyx.gpars.dataflow.DataflowBroadcast (DataflowBroadcast around DataflowStream[?])</p>
  </blockquote>
  <h2>Semi-pseudo code</h2>
  <p><strong>Workflow</strong></p>
  <pre><code>workflow my_WF {
      take:
          sample_input_file_ch
          original_ref
  
      main:
          GENERATE_FASTQS_PROC(sample_input_file_ch, original_ref)
          
          reads_per_job = 1_000_000
          fq_split_reads_and_RG_ch = GENERATE_FASTQS_PROC.out.fastqs
                          .splitFastq(by: reads_per_job, pe: true, file: true)
                          .map {
                              fq1, fq2 -&gt;
                                  tuple( fq1, fq2, GENERATE_FASTQS_PROC.out.sample_RG )
                          }
                          .view()
  
          ALIGN_FASTQ_BWA_PROC(fq_split_reads_and_RG_ch, ...)
  }
  </code></pre>
  <p><strong>GENERATE_FASTQS_PROC</strong></p>
  <pre><code>process GENERATE_FASTQS_PROC {
  
      input:
          path(sample_input_file)
          val(original_ref)
  
      output:
          tuple path(&quot;**/*_R1.fastq&quot;), path(&quot;**/*_R2.fastq&quot;), emit: fastqs
          path(&quot;**/*_RG.txt&quot;), emit: sample_RG
  
      script:
      &quot;&quot;&quot;
      generate_fastqs.sh <span class="math-container">$sample_input_file $</span>original_ref <span class="math-container">$task.cpus $</span>mem_per_thread
      &quot;&quot;&quot;
  }
  </code></pre>
  <p><strong>ALIGN_FASTQ_BWA_PROC</strong></p>
  <pre><code>process ALIGN_FASTQ_BWA_PROC {
  
      /*
       * Publish results. 'mode: copy' will copy the files into the publishDir
       * rather than only making links.
       */
      // publishDir(&quot;${params.results_dir}/01-REALIGN_SAMPLES&quot;, mode: 'copy')
  
      label 'REALIGN_SAMPLES'
  
      input:
          tuple path(fq1), path(fq2), path(sample_RG_file)
          val(align_to_ref)
          val(align_to_ref_tag)
          val(original_ref)
          val(output_format)
  
  
      output:
          path '*.{bam,cram}', emit: final_alignment
  
      script:
  
      /*
       * Tried to read the RG file using Groovy, but
       * got error that the file didn't exist.
       *
       * Does Nextflow not stage files until the actual .bash script starts?
       */
  //    println &quot;sample RG file: $sample_RG_file&quot;
  //    file_lines = sample_RG_file.readLines()
  //    sample_name = file_lines[0]
  //    sample_RG = file_lines[1]
  
      &quot;&quot;&quot;
      lines=()
      while read line
      do
          lines+=(&quot;\\\$line&quot;)
      done &lt; $sample_RG_file
  
      sample_name=&quot;\\\${lines[0]}&quot;
      sample_RG=&quot;\\\${lines[1]}&quot;
  
      bash realign_bwa-split.sh <span class="math-container">$fq1 $</span>fq2 \\\$sample_name \&quot;\\\$sample_RG\&quot; <span class="math-container">${align_to_ref} $</span>{output_format} $task.cpus
      &quot;&quot;&quot;
  }
  </code></pre>
  <hr />
  <h1>Update</h1>
  <p>Per @Steve's suggestion, I'm trying to have <code>GENERATE_FASTQS_PROC</code> create the split <code>.fastq</code> files using <code>Unix</code> <code>split</code>, but I'm still struggling with various concepts.</p>
  <p>Supposed I have this emitted from my process:</p>
  <pre><code>[sample_A_RG.txt, [s1_R1.split_00.fq, s1_R2.split_00.fq, ..., s1_R1.split_XX.fq, s1_R2.split_XX.fq]]`
  </code></pre>
  <p>In my mind, I want to pull out <code>s1_RG.txt</code> file from the <code>Channel</code> and extract the <code>@RG</code>. Then I want to call:</p>
  <pre><code>[&lt;all_fastqs_for_sample_A&gt;]
      .map{ 
          // pull out split ID numbers 
      }
      .groupTuple()
  </code></pre>
  <p>I believe this would group <code>_R1</code> and <code>_R2</code> for each split set, and then call</p>
  <pre><code>[[00.fq1, 00.fq2], ..., [XX.fq1, XX.fq2]]
      .combine(sample_RG_var)
  </code></pre>
  <p>to get <code>[[@RG, 00.fq1, 00.fq2], ..., [@RG, XX.fq1, XX.fq2]]</code>.</p>
  <p>Seems like it would be so much easier if I could use <code>Channel</code> factories and operators (e.g., <code>.fromFilePairs().combine( rg )</code>) in the <code>output:</code>.</p>
  <p><em>Getting files into a <code>process</code> is relatively straightforward, but getting them out in a reasonable manner is less so.</em></p>
  
Tags: <nextflow>
Title: Nextflow (DSL v2): how to best synchronize multiple outputs from a single process
ThreadText: # Nextflow (DSL v2): how to best synchronize multiple outputs from a single process
  
  (Asked by: username_0 on 2022-02-11)
  
  I have a workflow that needs to:
  
  1. Generate `.fastq` files from `.bams` (while preserving the `@RG` group from the original bam)
  2. Split the `.fastqs`
  3. Align
  4. etc.
  
  The only way I could think to preserve the `@RG` group was to print it to a file during step 1 and `emit` the file. I can't figure out how to synchronize the emitted `@RG` file with the `.fastqs`, however. I've tried several variations without luck.
  
  Questions:
  
  1. How do I synchronize the emitted `@RG` file (`sample_RG_file`) with the split `.fastqs`?
  2. Is there a better way?
  3. **Bonus question** (may be worth a separate post): does Nextflow not stage files until the actual .bash script starts? I tried to operate on the `sample_RG_file` within the `script` using `Groovy`, but got an error that the file didn't exist. (The code was working before I found I needed to synchronize `sample_RG_file` with the
  
  Other variations:
  
  1. The code was 'working' before I found I needed to synchronize `sample_RG_file` with the split `.fastqs`. I was originally passing `GENERATE_FASTQS_PROC.out.sample_RG` as a separate parameter to `ALIGN_FASTQ_BWA_PROC`, but it was always passing the same (first) file.
  2. I also tried `GENERATE_FASTQS_PROC.out.sample_RG.flatten()` in the `.map` operator, but that had the same error.
  
  **Current error:**
  
  The code below fails with the following error:
  
  > Error executing process > 'REALIGN\_SAMPLES\_WF:ALIGN\_FASTQ\_BWA\_PROC (1)'
  
  > Caused by:
  > Not a valid path value type: groovyx.gpars.dataflow.DataflowBroadcast (DataflowBroadcast around DataflowStream[?])
  
  ## Semi-pseudo code
  
  **Workflow**
  
  ```
  workflow my_WF {
      take:
          sample_input_file_ch
          original_ref
  
      main:
          GENERATE_FASTQS_PROC(sample_input_file_ch, original_ref)
          
          reads_per_job = 1_000_000
          fq_split_reads_and_RG_ch = GENERATE_FASTQS_PROC.out.fastqs
                          .splitFastq(by: reads_per_job, pe: true, file: true)
                          .map {
                              fq1, fq2 ->
                                  tuple( fq1, fq2, GENERATE_FASTQS_PROC.out.sample_RG )
                          }
                          .view()
  
          ALIGN_FASTQ_BWA_PROC(fq_split_reads_and_RG_ch, ...)
  }
  
  ```
  
  **GENERATE\_FASTQS\_PROC**
  
  ```
  process GENERATE_FASTQS_PROC {
  
      input:
          path(sample_input_file)
          val(original_ref)
  
      output:
          tuple path("**/*_R1.fastq"), path("**/*_R2.fastq"), emit: fastqs
          path("**/*_RG.txt"), emit: sample_RG
  
      script:
      """
      generate_fastqs.sh $sample_input_file $original_ref $task.cpus $mem_per_thread
      """
  }
  
  ```
  
  **ALIGN\_FASTQ\_BWA\_PROC**
  
  ```
  process ALIGN_FASTQ_BWA_PROC {
  
      /*
       * Publish results. 'mode: copy' will copy the files into the publishDir
       * rather than only making links.
       */
      // publishDir("${params.results_dir}/01-REALIGN_SAMPLES", mode: 'copy')
  
      label 'REALIGN_SAMPLES'
  
      input:
          tuple path(fq1), path(fq2), path(sample_RG_file)
          val(align_to_ref)
          val(align_to_ref_tag)
          val(original_ref)
          val(output_format)
  
      output:
          path '*.{bam,cram}', emit: final_alignment
  
      script:
  
      /*
       * Tried to read the RG file using Groovy, but
       * got error that the file didn't exist.
       *
       * Does Nextflow not stage files until the actual .bash script starts?
       */
  //    println "sample RG file: $sample_RG_file"
  //    file_lines = sample_RG_file.readLines()
  //    sample_name = file_lines[0]
  //    sample_RG = file_lines[1]
  
      """
      lines=()
      while read line
      do
          lines+=("\\\$line")
      done < $sample_RG_file
  
      sample_name="\\\${lines[0]}"
      sample_RG="\\\${lines[1]}"
  
      bash realign_bwa-split.sh $fq1 $fq2 \\\$sample_name \"\\\$sample_RG\" ${align_to_ref} ${output_format} $task.cpus
      """
  }
  
  ```
  
  ---
  
  # Update
  
  Per @username_1's suggestion, I'm trying to have `GENERATE_FASTQS_PROC` create the split `.fastq` files using `Unix` `split`, but I'm still struggling with various concepts.
  
  Supposed I have this emitted from my process:
  
  ```
  [sample_A_RG.txt, [s1_R1.split_00.fq, s1_R2.split_00.fq, ..., s1_R1.split_XX.fq, s1_R2.split_XX.fq]]`
  
  ```
  
  In my mind, I want to pull out `s1_RG.txt` file from the `Channel` and extract the `@RG`. Then I want to call:
  
  ```
  [<all_fastqs_for_sample_A>]
      .map{ 
          // pull out split ID numbers 
      }
      .groupTuple()
  
  ```
  
  I believe this would group `_R1` and `_R2` for each split set, and then call
  
  ```
  [[00.fq1, 00.fq2], ..., [XX.fq1, XX.fq2]]
      .combine(sample_RG_var)
  
  ```
  
  to get `[[@RG, 00.fq1, 00.fq2], ..., [@RG, XX.fq1, XX.fq2]]`.
  
  Seems like it would be so much easier if I could use `Channel` factories and operators (e.g., `.fromFilePairs().combine( rg )`) in the `output:`.
  
  *Getting files into a `process` is relatively straightforward, but getting them out in a reasonable manner is less so.*
  
  **Tags:** nextflow
  
  
  ---
  
  ## 2 answers
  
  ### Answer by username_1 on 2022-02-11. Score: 4 | Accepted answer
  
  Nextflow channels guarantee that items are delivered in the same order as they are sent. So if a process declares two (or more) output channels, the items that are emitted should already be synchronized. This is true unless of course one of the channels sets the [optional output](https://www.nextflow.io/docs/latest/process.html#optional-output) attribute.
  
  > 1. How do I synchronize the emitted `@RG` file (`sample_RG_file`) with the
  >    split `.fastqs`?
  
  Since the [splitFastq](https://www.nextflow.io/docs/latest/operator.html#splitfastq) operator *can* create *n* outputs for each input, you need some way to [combine](https://www.nextflow.io/docs/latest/operator.html#combine) the @RG files using some key. This key, for example, could be the basename of the sample input file. This, I think, would work nicely given the input file should just have a '.bam' extension. Another option, would be to supply some other value in the input declaration and just use that in your output declaration. If we went with the former, your output declaration might look like:
  
  ```
  output:
  tuple val(sample_input_file.baseName), path("sample_R1.fastq"), path("sample_R2.fastq"), \
      emit: fastqs
  tuple val(sample_input_file.baseName), path("sample_RG.txt"), \
      emit: sample_RG
  
  ```
  
  Then, the body of your workflow might look like:
  
  ```
  workflow {
  
      reads = GENERATE_FASTQS_PROC(sample_input_file_ch, original_ref)
          
      reads.fastqs
          .splitFastq( by: params.reads_per_job, pe: true, file: true )
          .combine( reads.sample_RG, by: 0 )
          .map { sample, fq1, fq2, rg_txt -> tuple(fq1, fq2, rg_txt) }
          .set { fq_split_reads_and_RG_ch }
  
      ...
  }
  
  ```
  
  You'll notice that I removed your fuzzy glob patterns from your output declaration above. Your workflow definition suggests that the `generate_fastqs.sh` script produces a single pair of FASTQ files and a single readgroup file for a given set of inputs. However, as you know, glob patterns can match one or more output files. When declaring output files, it's always best to name these guys explicitly. If you do use a glob pattern and for some reason your process was to produce two files that match a given pattern, you'll instead get back a tuple, which would break the pipeline.
  
  > 2. Is there a better way?
  
  Since you need the @RG tags with the split FASTQ files, I don't think there is. However, I generally avoid using the [splitting](https://www.nextflow.io/docs/latest/operator.html#splitting-operators) operators on large files. My preference is always to pass "work" off (including IO) to a separate process. A separate process can also provide increased flexibility if/when requirements change.
  
  > 3. Does Nextflow not stage files until the actual .bash script starts? I tried to operate
  >    on the sample\_RG\_file within the script using Groovy, but got an error
  >    that the file didn't exist.
  
  Nextflow evaluates the code in the script definition before returning the script string to be executed. This is why it's possible to implement a [conditional script](https://www.nextflow.io/docs/latest/process.html#conditional-scripts). Basically, you want to avoid all IO operations here. The script string to be executed has not even been decided at this point.
  
  ---
  
  **Update**
  
  I'm not sure how you plan to operate per sample. IIUC, `[<all_fastqs_for_sample_A>]` is basically a Channel of lists, where each list contains all split FASTQ files for a given sample. Using [map](https://www.nextflow.io/docs/latest/operator.html#map) here to pull out the split ID numbers followed by the [groupTuple](https://www.nextflow.io/docs/latest/operator.html#grouptuple) operator might not give the expected results. You might also run into the problems with calling [groupTuple](https://www.nextflow.io/docs/latest/operator.html#grouptuple), since, without a `size` attribute, it will basically wait for all outputs before emitting the grouped result:
  
  > You should always specify the number of expected elements in each
  > tuple using the `size` attribute to allow the `groupTuple` operator to
  > stream the collected values as soon as possible. However there are use
  > cases in which each tuple has a different size depending on the
  > grouping key. In this case use the built-in function `groupKey` that
  > allows you to create a special grouping key object to which it’s
  > possible to associate the group size for a given key.
  
  Part of the problem is that we don't know what the `generate_fastqs.sh` script does and how the other scripts work. If you are aligning using BWA MEM and have [samtools](http://www.htslib.org/doc/samtools.html) available in your $PATH, it might even be easier to produce interleaved output (since BWA MEM can read interleaved FASTQs) using the [samtools fastq](http://www.htslib.org/doc/samtools-fasta.html) command. This would save some work having to group each list of FASTQs into pairs, ensuring they are ordered as expected. This would be reasonably involved.
  
  Not sure what BAMs you need to process, or if you're looking for a more generic solution, but I thought I'd throw in an alternative solution you might find works well for you or that you may not have yet considered. Since BAMs often contain multiple readgroups, I find that if I split by readgroup (using [samtools split](http://www.htslib.org/doc/samtools-split.html), [flatten](https://www.nextflow.io/docs/latest/operator.html#flatten) the output Channel, then [samtools collate](http://www.htslib.org/doc/samtools-collate.html) followed by [samtools fastq](http://www.htslib.org/doc/samtools-fasta.html), I can achieve somewhat uniform mapping walltimes (assuming the input BAMs are similar) without needing to chunk the reads and then merge the alignments back together. If your input BAMs are not similar at all, then you will likely need to chunk to generate uniform alignment jobs.
  
  Below is an example, that shows how to create a groupKey, followed by the transpose, join/combine and groupTuple pattern. There are some caveats and there is some room for optimization:
  
  ```
  nextflow.enable.dsl=2
  
  params.original_ref = './ref.fasta'
  params.reads_per_job = 10000
  
  process samtools_view_header {
  
      tag { "${sample_name}:${bam.name}" }
  
      cpus 1
      memory 1.GB
  
      input:
      tuple val(sample_name), path(bam)
  
      output:
      tuple val(sample_name), path("${bam.baseName}.header.txt")
  
      """
      samtools view \\
          -H \\
          -o "${bam.baseName}.header.txt" \\
          "${bam}"
      """
  }
  
  process samtools_name_sort {
  
      tag { "${sample_name}:${bam.name}" }
  
      cpus 4
      memory 8.GB
  
      input:
      tuple val(sample_name), path(bam)
  
      output:
      tuple val(sample_name), path("${bam.baseName}.nsorted.bam")
  
      script:
      def avail_mem = task.memory ? task.memory.toGiga().intdiv(task.cpus) : 0
      def mem_per_thread = avail_mem ? "-m ${avail_mem}G" : ''
  
      """
      samtools sort \\
          -@ "${task.cpus - 1}" \\
          ${mem_per_thread} \\
          -n \\
          -o "${bam.baseName}.nsorted.bam" \\
          -T "${bam.baseName}.nsorted" \\
          "${bam}"
      """
  }
  
  process samtools_fastq {
  
      tag { "${sample_name}:${bam.name}" }
  
      cpus 1
      memory 2.GB
  
      input:
      tuple val(sample_name), path(bam)
  
      output:
      tuple val(sample_name), path("${bam.baseName}.fastq")
  
      """
      samtools fastq \\
          -O \\
          -T RG,BC \\
          -0 /dev/null \\
          "${bam}" \\
          > \\
          "${bam.baseName}.fastq"
      """
  }
  
  process split_fastq {
  
      tag { "${sample_name}:${fastq.name}" }
  
      cpus 1
      memory 1.GB
  
      input:
      tuple val(sample_name), path(fastq)
  
      output:
      tuple val(sample_name), path("${fastq.baseName}.${/[0-9]/*5}.fastq")
  
      """
      split \\
          --suffix-length=5 \\
          --additional-suffix=".fastq" \\
          -d \\
          -l "${params.reads_per_job*4}" \\
          "${fastq}" \\
          "${fastq.baseName}."
      """
  }
  
  process bwa_index {
  
      tag { fasta.name }
  
      cpus 1
      memory 12.GB
  
      input:
      path fasta
  
      output:
      tuple val(fasta.name), path("${fasta}.{amb,ann,bwt,pac,sa}")
  
      """
      bwa index "${fasta}"
      """
  }
  
  process bwa_mem {
  
      tag { "${sample_name}:${fastq.name}" }
  
      cpus 8
      memory 12.GB
  
      input:
      tuple val(sample_name), path(fastq), path(header)
      tuple val(idxbase), path(bwa_index)
  
      output:
      tuple val(sample_name), path("${fastq.baseName}.bam")
  
      script:
      def task_cpus = task.cpus > 1 ? task.cpus - 1 : task.cpus
  
      """
      bwa mem \\
          -p \\
          -t ${task_cpus} \\
          -C \\
          -H <(grep "^@RG" "${header}") \\
          "${idxbase}" \\
          "${fastq}" |
      samtools view \\
          -1 \\
          -o "${fastq.baseName}.bam" \\
          -
      """
  }
  
  process samtools_coordinate_sort {
  
      tag { "${sample_name}:${bam.name}" }
  
      cpus 4
      memory 8.GB
  
      input:
      tuple val(sample_name), path(bam)
  
      output:
      tuple val(sample_name), path("${bam.baseName}.csorted.bam")
  
      script:
      def avail_mem = task.memory ? task.memory.toGiga().intdiv(task.cpus) : 0
      def mem_per_thread = avail_mem ? "-m ${avail_mem}G" : ''
  
      """
      samtools sort \\
          -@ "${task.cpus - 1}" \\
          ${mem_per_thread} \\
          -o "${bam.baseName}.csorted.bam" \\
          -T "${bam.baseName}.csorted" \\
          "${bam}"
      """
  }
  
  process samtools_merge {
  
      tag { sample_name }
  
      cpus 1
      memory 4.GB
  
      input:
      tuple val(sample_name), path(bams)
  
      output:
      tuple val(sample_name), path("${sample_name}.bam{,.bai}")
  
      """
      samtools merge \\
          -c \\
          -p \\
          "${sample_name}.bam" \\
          ${bams}
      samtools index \\
          "${sample_name}.bam"
      """
  }
  
  workflow {
  
      original_ref = file( params.original_ref )
      bwa_index( original_ref )
  
      Channel.fromPath('./data/*.bam')
          | map { tuple( it.baseName, it ) }
          | set { sample_bam_files }
  
      samtools_view_header( sample_bam_files )
  
      sample_bam_files
          | samtools_name_sort
          | samtools_fastq
          | split_fastq
          | map { sample_name, fastq_files ->
              tuple( groupKey(sample_name, fastq_files.size()), fastq_files )
          }
          | transpose()
          | combine( samtools_view_header.out, by: 0 )
          | set { realignment_inputs }
  
      bwa_mem( realignment_inputs, bwa_index.out )
          | samtools_coordinate_sort
          | groupTuple()
          | map { sample_name, bam_files ->
              tuple( sample_name.toString(), bam_files )
          }
          | samtools_merge
          | view()
  }
  
  ```
  
  There are some caveats however, in no particular order:
  
  * It unravels your shell scripts. I think this is a better solution though because it gives you control over each of the components.
  * No CRAM support. But this should be easy to add if needed.
  * A single process to read a BAM header is wasteful, especially if the input BAM is large. More time will be spent localizing the file than actually doing work.
  * Replace `samtools sort -n` with `samtools collate`. There's really no need to name sort. You may prefer to pipe the output of `samtools collate` into `samtools fastq` for avoid writing an extra file. This might be useful if you need to operate on an extremely large number of large BAMs.
  * The 'samtools\_fastq' step should write gzipped output with low compression. This means replacing coreutils `split` with something else. A solution using Biopython could be useful here to read and write gzipped output in a single pass.
  * The last 'map' command in the workflow is unnecessary. I just thought I'd show that calling `.toString()` on the GroupKey gives you back an actual String object.
  * Also, it's not necessary to pass around a tuple of 'sample name' and paths as process inputs and outputs. There's no harm in doing this though. When you are more confident with Nextflow Channels, refactoring these processes should be easy.
  * My `samtools sort` processes ensure that they do not oversubscribe the nodes that they land on. They use approximately 320% - 380% CPU, depending on the size of the BAM. The resource requests for the other processes are not optimized at all.
  * With newer versions of samtools, BAM indexing can be done at the same time as merging, with the `--write-index` option.
  
  > **Comments:**
  > **username_0**: Thanks, @username_1. You've made several important points. Question about your suggested `workflow` code: wouldn't `.splitFastq` fail since `reads.fastqs` contains a tuple including the `sample_file_input.baseName`? I've been battling how to work with the `tuples` of `tuples` in various scenarios (e.g., in this situation and after a `.groupTuple()`).  
  > **username_0**: Also, per your suggestion, I've been working to have `GENERATE_FASTQS_PROC` split the `.fastqs` itself and emit those. That's causing me a headache. The level of abstraction in `Nextflow` has really made these details quite elusive for me.  
  > **username_0**: I added an update to my question explaining what I feel like I should be able to do and what I ultimately want to get.  
  > **username_0**: Thanks, @username_1. I think the interleaved option might be the simplest. Great suggestion. Still, I'm shocked by how difficult it is to get files out of a `process` without mixing them between samples. What kills me is, I know all files for a single sample are safely contained in a unique directory, but I'm forced to mix them up with files from other samples in the same `Channel`, so I'm left with convoluted methods to try to distinguish them from each other. If I `emit` the `path` to the directories and then use `Channel.fromPath()`, it would be sooooo easy. But that's not possible, AFAIK.  
  > **username_0**: `generate_fastqs.sh` takes a single `.bam` -> sorts by name -> converts to `.fastq` (currently w/ `bedtools`), and splits them with `Unix` `split`. The files are stored safe and sound, all tidy-like in a unique directory, where I have clear control over what's generated and mixed together (unlike `Nextflow` `Channels`). Forgive the frustration. :-)  
  > **username_1**: Sorry @username_0, I was meant to let you know yesterday that I had updated my answer but got pulled in another direction before I could finish the update. WRT your first question above, [splitFastq](https://www.nextflow.io/docs/latest/operator.html#splitfastq) will work fine with a tuple that includes a sample identifier. For example, it will work with the output of [fromFilePairs](https://www.nextflow.io/docs/latest/channel.html#fromfilepairs) with `flat: true`.  
  > **username_1**: @username_0 I struggled with the same thing when I first started with NF. I found that outputting tuples of files, prefixed with some identifier (usually a sample name) helps considerably when working with multiple channels.  
  > **username_0**: This was a huge help, @username_1. You've also made several additional helpful suggestions. It had never dawned on me that `samtools` was oversubscribing because `-@` is for *additional* threads, though that should have been completely obvious. Really appreciate your help. I'm still not a fan of NF `output` `Channels`, but maybe I'll get there. :-)  
  
  ---
  
  ### Answer by username_0 on 2022-02-13. Score: 3
  
  # Update 3
  
  While the code in **Update 2** works, @username_1 pointed out that my approach may not be ideal in cloud environments (see comments). He also provided a `Nextflow` solution that works. I'm still not a fan of how `Nextflow` `output:` `Channels` work, but that may be from my inexperience.
  
  # Update 2
  
  Pretty sure I got it this time. **The following code works**. I'm getting all of the `tuples` I was expecting.
  
  Here's the updated code for my `workflow` to create the tuples from each line in the `sample_tuple_file` I created from a `bash` `for` loop. I spent quite a while trying to get a combination of `.splitText()` and `.splitCSV()` to work, but for some reason, you cannot chain as follows: `it.splitText().splitCSV()`.
  
  ```
       GENERATE_FASTQS_PROC.out.sample_tuple_file
           .map {
              it.splitText()
           }
           .flatten()
           .map {
               toks = it.trim().split(',')
               sample_name = toks[0]
               fq = toks[1]
               rg = toks[2]
               tuple(sample_name, fq, rg)
               }
           .view()
          .set { sample_fq_tuples_ch }
  
  ```
  
  ---
  
  # Update 1
  
  I thought I had it, but I'm actually only getting ONE of the tuples from my `GENERATE_FASTQS_PROC.out.sample_tuple_file.map { }` statement (the last one). And I have no idea why...
  
  ---
  
  # Original posted answer
  
  Despite @username_1's exceptional help and patience with me, I have really battled to figure out how to work with the `Nextflow` `output:` `Channels`, but simply could not figure out any solution using them.
  
  I finally figured out a super simple method that is much easier than dealing with `Nextflow` `Channels`, which I share here in case it will help others. I personally will probably use this method from now on, unless someone can convince me that's a bad idea. In my view, using my approach gives me complete control over the way outputs and inputs are handled, and I don't have to worry whether `Nextflow` is doing what I hope it's doing.
  
  **generate\_fastqs.sh**
  `generate_fastqs.sh` takes a single `.bam` -> sorts by name -> converts to .fastq (now w/ `samtools fastq`, thanks to @username_1), and splits them with `Unix` `split`. Then, at the end of the script, I create a single file with all of the `tuples` that I want to pass to my next `process`.
  
  Here are the key lines for these steps (including exit codes to ensure `Nextflow` receives error notifications). The last step is the key (for me, anyway):
  
  ```
  ...
  
  #####################
  # Sort .bam by name #
  #####################
  if ! time samtools sort -@ $n_threads -n -m ${mem_per_thread}G $sample_input > $sample_sorted_by_name; then
      echo "ERROR: sort failed"
      exit 1
  fi
  
  ...
  
  #############################
  # Create interleaved .fastq #
  #############################
  if ! time samtools fastq -o $fq -s /dev/null -0 /dev/null --reference $original_ref $sample_sorted_by_name; then
      echo "ERROR: bamtofastq failed"
      exit 1
  fi
  
  ...
  
  ################
  # Split fastqs #
  ################
  if ! time split --additional-suffix=".fastq" -dl $n_lines $fq "${split_dir}/${sample_name}.interleaved_R1_R2.split."; then
      echo "ERROR: split failed"
      exit 1
  fi
  
  ...
  
  ###########################################################
  # Store sample name, .fastq, and RG to file for alignment #
  ###########################################################
  RG_file="${sample_name}.tuples.txt"
  
  # Create empty file
  > $RG_file
  for fq_file in $PWD/${split_dir}/*
  do
  echo "${sample_name},$fq_file,$RG" >> $RG_file    # <-- Creating future tuples in file, here (comma separated).
  done
  
  ```
  
  **GENERATE\_FASTQS\_PROC**
  
  In `GENERATE_FASTQS_PROC`, `emit` one simple file--the tuple file I created myself with a simple `bash` loop (so easy and feels so good):
  
  ```
      emit:
          path("*.tuples.txt"), emit: sample_tuple_file
  
  ```
  
  **Workflow**
  
  In my `workflow`, I call `GENERATE_FASTQS_PROC`, access the file from the `Channel`, read each line, and create a `Nextflow` tuple for each line (again, the lines I so easily created in a `bash` `for` loop).
  
  ```
      /*
       * Generate fastq files from the original .(cr|b)am file and split into
       * mini .fastq files with `n_reads_per_run` in each mini .fastq
       */
      GENERATE_FASTQS_PROC(sample_input_file_ch, n_reads_per_run, original_ref)
  
      /*
       * Get sample name, fastq, and read group (RG) from the tuple file using Nextflow
       */
  
       GENERATE_FASTQS_PROC.out.sample_tuple_file
          .map{
              sample_tuple_file ->
                  sample_tuple_file.withReader {          // Open file
                      it.eachLine {                       // Loop over each line in file
                          line ->
                              toks = line.split(',')      // Split by comma
                              sample_name = toks[0]       // Get each item
                              fq = toks[1]
                              rg = toks[2]
  
                              tuple(sample_name, fq, rg)  // Create the simple tuple that has eluded me for two days!
                      }
                  }
          }
          .set { sample_fq_tuples_ch }                    // Assign it to a new Channel name and move along!
  
      /*
       * Align fastq file pairs
       */
  
      ALIGN_FASTQ_BWA_PROC(sample_fq_tuples_ch, align_to_ref,
                               align_to_ref_tag, original_ref, output_format)
  ```
  
  ```
  
  > **Comments:**
  > **username_0**: Ugh. I thought I had it, but I confused myself with my own print statements. I'm actually only getting ONE of the tuples from my `GENERATE_FASTQS_PROC.out.sample_tuple_file.map { }` statement. And I have no idea why...  
  > **username_0**: Pretty sure I got it working with update #2. Hopefully others can learn from my mistakes and final victory.  
  > **username_1**: I can see you have put in quite a bit of effort here, but this is not something I would recommend to others. This is a solution that will work on a local filesystem _only_, where all files exist under a common root. This is what makes it possible to pick them up later on if you have a pointer to them (your tuples.txt contains file pointers). Since the required files are not declared in the output block, Nextflow will not attempt to "unstage" them (upload them to a remote object store, like a bucket) when you deploy your workflow to the cloud. They will just be lost.  
  > **username_1**: Thanks for including your `generate_fastqs.sh`. I have a better picture now of what you are trying to do. I will try to update my answer with a working solution soon. I just need to grab a couple of test BAMs. I should have a couple somewhere..  
  > **username_0**: True. I hadn’t considered cloud file systems. I was worried you’d burst my bubble. :-)  
  > **username_0**: Couldn’t I still emit the files you mention, but use my method to generate tuples, etc? Or, couldn’t I publish results and access them that way? I realize this breaks convention, but it appears that, while Nextflow has many advantages,  Nextflow’s output Channels are fundamentally convoluted, flawed, and maybe even dangerous. I’d love to be proven wrong. I don’t know Nextflow’s internals, but seems like this could be easily fixed. e.g., if every process instance assigned an ID to the process that we could filter by upon return, I could trust that NF is giving me what I think I’m getting.  
  > **username_1**: You could perhaps. But that would not be a pleasant experience at all and would produce a convoluted solution. I don't have any real problem with Nextflow Channels. I think they're great and they make writing complex workflows possible. I've updated my answer above. I think it's ok as an example but there are some caveats listed. You may or may not need to address these. But it will likely need some tinkering before you can run with it.  
  
  ---
  
  