<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>

CC BY-SA 4.0
2022-02-13T01:51:18.667Z
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...
Mark Ebbert
CC BY-SA 4.0
2022-02-13T03:11:50.917Z
Pretty sure I got it working with update #2. Hopefully others can learn from my mistakes and final victory.
Mark Ebbert
CC BY-SA 4.0
2022-02-13T14:52:23.093Z
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.
Steve
CC BY-SA 4.0
2022-02-13T15:02:10.017Z
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..
Steve
CC BY-SA 4.0
2022-02-14T02:56:51.773Z
True. I hadn’t considered cloud file systems. I was worried you’d burst my bubble. :-)
Mark Ebbert
CC BY-SA 4.0
2022-02-14T03:10:28.790Z
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.
Mark Ebbert
CC BY-SA 4.0
2022-02-15T06:33:42.097Z
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.
Steve
CC BY-SA 4.0
2022-02-13T01:24:35.137Z
2022-02-16T15:02:17.650Z
2022-02-16T15:02:17.650Z
Mark Ebbert
<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>

CC BY-SA 4.0
2022-02-12T02:46:19.437Z
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()`).
Mark Ebbert
CC BY-SA 4.0
2022-02-12T02:47:57.633Z
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.
Mark Ebbert
CC BY-SA 4.0
2022-02-12T03:43:26.473Z
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.
Mark Ebbert
CC BY-SA 4.0
2022-02-12T21:31:41.087Z
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.
Mark Ebbert
CC BY-SA 4.0
2022-02-12T21:33:44.030Z
`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. :-)
Mark Ebbert
CC BY-SA 4.0
2022-02-13T13:37:01.680Z
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`.
Steve
CC BY-SA 4.0
2022-02-13T14:12:39.217Z
@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.
Steve
CC BY-SA 4.0
2022-02-16T14:57:37.613Z
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. :-)
Mark Ebbert
CC BY-SA 4.0
2022-02-11T16:54:55.763Z
2022-02-15T06:34:44.010Z
2022-02-15T06:34:44.010Z
Steve
<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>

<nextflow>
Nextflow (DSL v2): how to best synchronize multiple outputs from a single process
# 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:

Generate .fastq files from .bams (while preserving the @RG group from the original bam)
Split the .fastqs
Align
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:

How do I synchronize the emitted @RG file (sample_RG_file) with the split .fastqs?
Is there a better way?
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:

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.
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 @Steve'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:
[]
    .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 attribute.


How do I synchronize the emitted @RG file (sample_RG_file) with the
split .fastqs?


Since the splitFastq operator can create n outputs for each input, you need some way to 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.


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 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.


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. 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, [] is basically a Channel of lists, where each list contains all split FASTQ files for a given sample. Using map here to pull out the split ID numbers followed by the groupTuple operator might not give the expected results. You might also run into the problems with calling 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 available in your $PATH, it might even be easier to produce interleaved output (since BWA MEM can read interleaved FASTQs) using the samtools fastq 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, flatten the output Channel, then samtools collate followed by samtools fastq, 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, @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()`).  
> **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, @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.  
> **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 @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`.  
> **username_1**: @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.  
> **username_0**: 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. :-)  

---

### Answer by username_0 on 2022-02-13. Score: 3

Update 3
While the code in Update 2 works, @Steve 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 @Steve'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 @Steve), 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.  

---

