1
$\begingroup$

This is my first nextflow pipeline and i am struggling with the .fromFilePairs concept of Nextflow.

I want to process paired-end read files and align them against a reference genome. I use Hisat2 so the index files come in form of 8 files with the suffix .ht2. Hisat needs the file path and the filename without ".ht2". All worked well with a single-end version of the process. The reference also has to be passed through a channel because i use a hisat docker-container and otherwise the files are not mounted.

The idea is to pass read-pairs to a preprocessing step and then pass them (as tuples, like a Channel.fromPairs()) into the hisat process and then further downstream.

My problem is all file-pairs are processed during the first step but only the first pair is passed further downstream. All downstream steps finish with the one pair and the rest simply stays in the output-folder of the preprocessing step.

I tried to simply skip the first step to see if the "connection" between the first and second process is not working but the Channel which works for the first step does not for the hisat step. So i think there is a problem with the combination of multiple read-pairs as input but only one reference file?!

process hisat2_paired {
  tag "$sample_id"
 
  container "${params.container_hisat}"
  publishDir "${params.out_dir}/Nextflow_output/hisat/"

  input:
  tuple val(sample_id), path(input_files)
  path( index_files )

  output:
  path('*.sam') , emit: hisat2_sam_out_ch
  path('*.txt')

  script:
  """
  INDEX=`find -L ${index_files}/ -name "*.1.ht2" | sed 's/.1.ht2//'` 
  hisat2 --rna-strandness RF \
          -x \\\$INDEX \
          -1 ${input_files[0]} \
      -2 ${input_files[1]} \
          -S ${sample_id}.sam &> ${sample_id}.hisat_summary.txt
  """

workflow {

Channel
    .fromFilePairs( "${params.reads_folder}/*_{R1,R2}*f*q.gz")
    .ifEmpty { exit 1, "no fastq files found at given path" }
    .view()
    .set{ read_pair_ch }

Channel
    .fromPath( params.hisat2_index )
    .ifEmpty { exit 1, "no hisat2 index files found - path was empty" }
    //.view()
    .set { hisat2_index_ch }

hisat2_paired( read_pair_ch, hisat2_index_ch )

}

I started the workflow with 3 test file-pairs and the Output on the CLI states that only one process is started although if i look at the Channel.view() all 3 file pairs are printed.

executor >  local (1)
[2f/30a4d6] process > hisat2_paired (MC3T3-Ko1_S2_L001) [  0%] 0 of 1
[MC3T3-Ko1_S2_L001, [/mnt/c/Path/to/MC3T3-Ko1_S2_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko1_S2_L001_R2_001.fastq.gz]]
[MC3T3-Ko2_S4_L001, [/mnt/c/Path/to/MC3T3-Ko2_S4_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko2_S4_L001_R2_001.fastq.gz]]
[MC3T3-Ko3_S6_L001, [/mnt/c/Path/to/MC3T3-Ko3_S6_L001_R1_001.fastq.gz, /mnt/c/Path/to/MC3T3-Ko3_S6_L001_R2_001.fastq.gz]]

I was already thinking about memory and resources. Since I prototype on my local PC i set the queue size in the config file to 2. I know that my hisat process needs about 7Gb RAM (for Mus.musculus) per process so there should be enough resources. Also the problem was that the local executor is not even trying to start a second process.

$\endgroup$
0

1 Answer 1

1
$\begingroup$

So i think there is a problem with the combination of multiple read-pairs as input but only one reference file?!

Correct. The problem is that the hisat2_index_ch is a queue channel, but what you want is a value channel so that the (file) values can be read an unlimited number of times without being consumed. Assuming params.hisat2_index is intended to supply a glob pattern, you could simply call the collect operator to produce a list of files as the sole emission:

Channel
    .fromPath( params.hisat2_index )
    .collect()
    .set { hisat2_index_ch }

You would then need to update your process definition accordingly. Alternatively, have your workflow create the HISAT2 index files for you. This simplifies things considerably, by requiring only a single reference FASTA file as input. Note that value channels are implicitly created as output for a process whose inputs are all value channels. This means in the example below, hisat2_index.out is also a value channel:

params.publish_dir = './results'

params.container_hisat = 'quay.io/biocontainers/hisat2:2.2.1--h87f3376_4'

params.reads = '*_R{1,2}.fastq.gz'
params.ref_fasta = 'chr1.fasta'

process hisat2_index {

    container params.container_hisat
    cpus 8

    input:
    path fasta

    output:
    tuple val(fasta.baseName), path("${fasta.baseName}.*.ht2")

    """
    hisat2-build \\
        -p ${task.cpus} \\
        "${fasta}" \\
        "${fasta.baseName}"
    """
}
process hisat2_paired {

    tag { sample_id }

    container params.container_hisat
    publishDir "${params.publish_dir}/hisat"

    input:
    tuple val(sample_id), path(input_files)
    tuple val(idx), path('hisat2-idx/*')

    output:
    path('*.sam'), emit: hisat2_sam
    path('*.txt')

    script:
    def (fq1, fq2) = input_files

    """
    hisat2 \\
      --rna-strandness RF \\
      -x "hisat2-idx/${idx}" \\
      -1 ${fq1} \\
      -2 ${fq2} \\
      -S ${sample_id}.sam \\
      &> ${sample_id}.hisat_summary.txt
    """
}
workflow {

    read_pairs = Channel.fromFilePairs( params.reads )
    ref_fasta = file( params.ref_fasta )

    hisat2_index( ref_fasta )

    hisat2_paired( read_pairs, hisat2_index.out )
}
$\endgroup$
5
  • $\begingroup$ Thank you for the help. I knew it was something simple. Simply adding the .collect() to the hisat_index_ch did the trick. But the process definition needed to be set as path or else the files are not available in the docker container. By "update the process definition" you meant changing the input from path() to val() right? $\endgroup$
    – stfn.snk
    Commented Nov 6, 2022 at 16:52
  • $\begingroup$ @stfn.snk no, sorry, I should have elaborated. Always use the path() qualifier for file input/outputs. The path() qualifier will work for a single file or a collection of files. I just meant that you will need to update how you define your HISAT2 INDEX variable. I don't think the find command will work as expected. It might be better to supply the basename of the index (i.e. a string) using val() in a tuple() like in the example above. $\endgroup$
    – Steve
    Commented Nov 7, 2022 at 2:32
  • $\begingroup$ One way, could be to use the map operator to transform the collection, similar to how the find command was intended to work, .e.g Channel.fromPath( params.hisat2_index ).collect().map { tuple( it.first().getBaseName(2), it ) }.set { ... } $\endgroup$
    – Steve
    Commented Nov 7, 2022 at 2:40
  • $\begingroup$ I tried to avoid the building process, since it needs considerable computing resources (so i've read) so I used a public available HISAT Index. I think i tried the .baseName function already but it gave me the index filename including the number and the process failed (e.g. genome.1.ht2 became genome.1) So i ended up with the find solution. I will look further into your suggestions. Thanks again, i really appreciate your help $\endgroup$
    – stfn.snk
    Commented Nov 7, 2022 at 22:26
  • $\begingroup$ @stfn.snk So you could use .getBaseName(2) to remove the double extension, or .simpleName if you can be sure the prefix doesn't contain a literal dot/period. Anytime! $\endgroup$
    – Steve
    Commented Nov 8, 2022 at 4:02

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.