You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Nextflow多菌株变异分析流程通道算子异常技术求助

avatar
开发者阿泽
2026-6-17
多菌株变异分析Pipeline异常问题

我正在开发多菌株变异分析Pipeline，遇到两个异常问题：

  • 使用Combine算子时，输入两个reads，但流程提示“Only one read is running all the steps”，最终只生成一个输出
  • 替换为Cross算子后，流程仅完成indexReference和trimReads两个步骤就停滞，后续分析步骤完全不执行

我的目标是实现多菌株输入支持，让每个菌株都能走完所有分析步骤并生成对应输出。相关代码及执行结果如下：

使用Combine算子时的代码与执行结果

代码 

#!/usr/bin/env nextflow

nextflow.enable.dsl=2

params.ref = ''    
params.reads = []
params.snpEff_db = 'Pseudomonas_aeruginosa'

def getFastqList() {
    def input = System.console().readLine('Enter fastq filenames separated by spaces: ')
    def inputList = input.split(' ') as List
    def fullPathList = inputList.collect { file(it).toAbsolutePath().toString() }
    return fullPathList
}
process indexReference {
    tag "INDEX:${params.ref}"

    input:
        path ref

    publishDir './ref_index/', mode: 'copy'

    output:
        path "${ref}*", emit: ref_index

    script:
        """
        bwa index $ref
        samtools faidx $ref
        """
}

process trimReads {
    tag "FASTP:${sampleid}"

    input:
        tuple val(sampleid), path(reads)

    publishDir './trimtry/', mode: 'copy'

    output:
        tuple val(sampleid), path("${sampleid}_trimmed.fastq.gz")  

    script:
        """
        fastp -i ${reads} -o ${sampleid}_trimmed.fastq.gz
        """
}

process alignReads {
    tag "BWA:${sampleid}"

    input:
        tuple val(sampleid), path(trimmed_reads), path(ref_index)

    publishDir './aligntry/', mode: 'copy'

    output:
        path "${sampleid}.bam", emit: aligned_bam

    script:
        """
        bwa mem ${params.ref} ${trimmed_reads} > ${sampleid}.sam
        samtools view -Sb ${sampleid}.sam > ${sampleid}.bam
        """
}
process sortBam {
    tag "SORT:${sampleid}"

    input:
        tuple val(sampleid), path(aligned_bam)

    publishDir './sorted_bam/', mode: 'copy'

    output:
        path "${sampleid}_sorted.bam", emit: sorted_bam

    script:
        """
        samtools sort -o ${sampleid}_sorted.bam ${aligned_bam}
        """
}

workflow {
    def reads = getFastqList()

    if (params.ref && reads) {
        println "Reference genome: ${params.ref}"
        println "Input reads: ${reads}"

        channel
            .fromPath(params.ref)
            | indexReference
            | set { ref_index }

        channel
            .from(reads)
            .map { file -> tuple(new File(file).getName().split('\\.')[0], file) }
            | trimReads
            | set { trim }
            
        trim
            .combine (ref_index)
            | alignReads
            |set {aligned}


        aligned
            .aligned_bam
            | map { bam -> tuple('aligned_bam', bam.toString()) }
            | sortBam
            | set { sorted }

        sorted
            .sorted_bam
            | map { bam -> tuple('sorted_bam', bam.toString()) }
            | removeDuplicates
            | set { dedup }

        dedup
            .dedup_bam
            | map { bam -> tuple('dedup_bam', bam.toString()) }
            | callVariants
            | set { variants }

        variants
            .vcf
            | map { vcf -> tuple(vcf.getName().split('\\.')[0], vcf) }
            | annotateVariants
            | set { annotated }

        annotated
            .annotated_vcf
            | map { vcf -> tuple(vcf.getName().split('\\.')[0], vcf) }
            | filterSNPs
            | set { filtered_snps }

        filtered_snps
            .view()
    } else {
        error "Please provide valid paths for the reference genome and input reads."
    }
}

执行结果 

(base) virudhagiri@virudhagiri-HP-Laptop-15s-eq2xxx:~/Desktop/ngs/helo$ nextflow run try14.nf --ref /home/virudhagiri/Desktop/ngs/sequence2.fasta

N E X T F L O W   ~  version 24.04.4
Launching try14.nf [stupefied_goldwasser] DSL2 - revision: 35d5b5e299
Enter fastq filenames separated by spaces: ERR010476.fastq SRR30404095.fastq
Reference genome: /home/virudhagiri/Desktop/ngs/sequence2.fasta
Input reads: [/home/virudhagiri/Desktop/ngs/helo/ERR010476.fastq, /home/virudhagiri/Desktop/ngs/helo/SRR30404095.fastq]
executor >  local (7)
[86/f08640] process > indexReference (INDEX:/home/virudhagiri/Desktop/ngs/sequence2.fasta) [100%] 1 of 1 ✔
[5f/848891] process > trimReads (FASTP:SRR30404095)                                        [ 50%] 1 of 2
executor >  local (8)
[86/f08640] process > indexReference (INDEX:/home/virudhagiri/Desktop/ngs/sequence2.fasta) [100%] 1 of 1 ✔
[5f/848891] process > trimReads (FASTP:SRR30404095)                                        [ 50%] 1 of 2
executor >  local (9)
[86/f08640] process > indexReference (INDEX:/home/virudhagiri/Desktop/ngs/sequence2.fasta) [100%] 1 of 1 ✔
[5f/848891] process > trimReads (FASTP:SRR30404095)                                        [ 50%] 1 of 2
executor >  local (10)
[86/f08640] process > indexReference (INDEX:/home/virudhagiri/Desktop/ngs/sequence2.fasta) [100%] 1 of 1 ✔
[eb/b4db34] process > trimReads (FASTP:ERR010476)                                          [100%] 2 of 2 ✔
executor >  local (15)
[86/f08640] process > indexReference (INDEX:/home/virudhagiri/Desktop/ngs/sequence2.fasta) [100%] 1 of 1 ✔
[eb/b4db34] process > trimReads (FASTP:ERR010476)                                          [100%] 2 of 2 ✔
[ef/5037a0] process > alignReads (BWA:ERR010476)                                           [100%] 2 of 2 ✔
[b7/f19097] process > sortBam (SORT:aligned_bam)                                           [100%] 2 of 2 ✔
[a8/47177c] process > removeDuplicates (DEDUP:sorted_bam)                                  [100%] 2 of 2 ✔
[e3/380746] process > callVariants (BCFTOOLS:dedup_bam)                                    [100%] 2 of 2 ✔
[a8/6680b8] process > annotateVariants (SNPEFF:dedup_bam)                                  [100%] 2 of 2 ✔
[2e/1a508e] process > filterSNPs (FILTER:dedup_bam_annotated)                              [100%] 2 of 2 ✔
/home/virudhagiri/Desktop/ngs/helo/work/2e/1a508e96d2bcd5c32dcbf2fdce8756/dedup_bam_annotated_filtered_snps.vcf
Completed at: 14-Oct-2024 14:13:53
Duration    : 2m 25s
CPU hours   : (a few seconds)
Succeeded   : 15

使用Cross算子时的代码与执行结果

代码 

#!/usr/bin/env nextflow

nextflow.enable.dsl=2

params.ref = ''    
params.reads = []
params.snpEff_db = 'Pseudomonas_aeruginosa'

def getFastqList() {
    def input = System.console().readLine('Enter fastq filenames separated by spaces: ')
    def inputList = input.split(' ') as List
    def fullPathList = inputList.collect { file(it).toAbsolutePath().toString() }
    return fullPathList
}
process indexReference {
    tag "INDEX:${params.ref}"

    input:
        path ref

    publishDir './ref_index/', mode: 'copy'

    output:
        path "${ref}*", emit: ref_index

    script:
        """
        bwa index $ref
        samtools faidx $ref
        """
}

process trimReads {
    tag "FASTP:${sampleid}"

    input:
        tuple val(sampleid), path(reads)

    publishDir './trimtry/', mode: 'copy'

    output:
        tuple val(sampleid), path("${sampleid}_trimmed.fastq.gz")  

    script:
        """
        fastp -i ${reads} -o ${sampleid}_trimmed.fastq.gz
        """
}

process alignReads {
    tag "BWA:${sampleid}"

    input:
        tuple val(sampleid), path(trimmed_reads), path(ref_index)

    publishDir './aligntry/', mode: 'copy'

    output:
        path "${sampleid}.bam", emit: aligned_bam

    script:
        """
        bwa mem ${params.ref} ${trimmed_reads} > ${sampleid}.sam
        samtools view -Sb ${sampleid}.sam > ${sampleid}.bam
        """
}
process sortBam {
    tag "SORT:${sampleid}"

    input:
        tuple val(sampleid), path(aligned_bam)

    publishDir './sorted_bam/', mode: 'copy'

    output:
        path "${sampleid}_sorted.bam", emit: sorted_bam

    script:
        """
        samtools sort -o ${sampleid}_sorted.bam ${aligned_bam}
        """
}

workflow {
    def reads = getFastqList()

    if (params.ref && reads) {
        println "Reference genome: ${params.ref}"
        println "Input reads: ${reads}"

        channel
            .fromPath(params.ref)
            | indexReference
            | set { ref_index }

        channel
            .from(reads)
            .map { file -> tuple(new File(file).getName().split('\\.')[0], file) }
            | trimReads
            | set { trim }
            
        trim
            .cross (ref_index)
            | alignReads
            |set {aligned}


        aligned
            .aligned_bam
            | map { bam -> tuple('aligned_bam', bam.toString()) }
            | sortBam
            | set { sorted }

        sorted
            .sorted_bam
            | map { bam -> tuple('sorted_bam', bam.toString()) }
            | removeDuplicates
            | set { dedup }

        dedup
            .dedup_bam
            | map { bam -> tuple('dedup_bam', bam.toString()) }
            | callVariants
            | set { variants }

        variants
            .vcf
            | map { vcf -> tuple(vcf.getName().split('\\.')[0], vcf) }
            | annotateVariants
            | set { annotated }

        annotated
            .annotated_vcf
            | map { vcf -> tuple(vcf.getName().split('\\.')[0], vcf) }
            | filterSNPs
            | set { filtered_snps }

        filtered_snps
            .view()
    } else {
        error "Please provide valid paths for the reference genome and input reads."
    }
}

执行结果 

(base) virudhagiri@virudhagiri-HP-Laptop-15s-eq2xxx:~/Desktop/ngs/helo$ nextflow run loop.nf --ref /home/virudhagiri/Desktop/ngs/helo/sequence2.fasta
N E X T F L O W   ~  version 24.04.4
Launching loop.nf [drunk_gutenberg] DSL2 - revision: d43dba08a0
Enter fastq filenames separated by spaces: ERR010476.fastq SRR30404095.fastq
Reference genome: /home/virudhagiri/Desktop/ngs/helo/sequence2.fasta
Input reads: [/home/virudhagiri/Desktop/ngs/helo/ERR010476.fastq, /home/virudhagiri/Desktop/ngs/helo/SRR30404095.fastq]
executor >  local (3)
[60/0cf263] process > indexReference (INDEX:/home/virudhagiri/Desktop/ngs/helo/sequence2.fasta) [100%] 1 of 1 ✔
[a6/be4ead] process > trimReads (FASTP:ERR010476)                                               [100%] 2 of 2 ✔
[-        ] process > alignReads                                                                -
[-        ] process > sortBam                                                                   -
[-        ] process > removeDuplicates                                                          -
[-        ] process > callVariants                                                              -
[-        ] process > annotateVariants                                                          -
[-        ] process > filterSNPs                                                                -

内容的提问来源于stack exchange，提问作者Virudhagiri E

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
火山引擎 最新活动
全面开放
Doubao-Seedance-2.5全新模型升级

模型再升级,30秒超长叙事, 模态参考扩容

HOT
方舟Codingplan

模型自由，工具不限，最新支持 Deepseek-V4 系列、GLM-5.3 系列

HOT
方舟Agent plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列

HOT
新用户特惠专场

大模型19元起，Al应用9.9元畅享，新人首购爆款尽享优惠

最近更新时间：2026.06.17 03:25:56