Nextflow多进程输出合并及后续进程输入配置问题

核心思路 #

利用Nextflow的join操作符，基于共同的pair_id将Process A（正常样本）和Process B（肿瘤样本）的输出通道自动配对，无需手动指定文件路径，批量处理所有样本。

步骤1：优化Process A和B的输出结构 #

将每个样本的BAM和索引文件打包为单个tuple输出，确保每个输出通道都携带pair_id作为匹配键：

process A {
  publishDir params.trim, mode: 'copy' 

  input:
  tuple val(pair_id), file(recal_readsbamfile)

  output:
  tuple val(pair_id), file(normal_samplefile), file(normal_samplefile_bai), emit: normal_output

  script:
  normal_samplefile = "${pair_id}_recal_reads_SM.bam"
  normal_samplefile_bai = "${pair_id}_recal_reads_SM.bai"

  """
  ./gatk --java-options "-Xmx24G" AddOrReplaceReadGroups \
    --INPUT $recal_readsbamfile \
    --OUTPUT $normal_samplefile \
    --RGLB lib1 --RGPL illumina --RGPU NONE --RGSM NORMAL_$pair_id
  ./gatk --java-options "-Xmx24G" BuildBamIndex --INPUT $normal_samplefile
  """
}

process B {
  publishDir params.trim, mode: 'copy' 

  input:
  tuple val(pair_id), file(recal_readsbamfile)

  output:
  tuple val(pair_id), file(tumor_samplefile), file(tumor_samplefile_bai), emit: tumor_output

  script:
  tumor_samplefile = "${pair_id}_recal_reads_SM.bam"
  tumor_samplefile_bai = "${pair_id}_recal_reads_SM.bai"

  """
  ./gatk --java-options "-Xmx24G" AddOrReplaceReadGroups \
    --INPUT $recal_readsbamfile \
    --OUTPUT $tumor_samplefile \
    --RGLB lib1 --RGPL illumina --RGPU NONE --RGSM TUMOR_$pair_id
  ./gatk --java-options "-Xmx24G" BuildBamIndex --INPUT $tumor_samplefile
  """
}

步骤2：合并通道并传入Process M #

通过join操作符按pair_id配对同一样本的正常和肿瘤BAM，再将结果传入Process M：

// 提取pair_id和BAM文件（若需传递索引可保留bai字段）
def normal_bams = A.out.normal_output.map { pair_id, bam, bai -> tuple(pair_id, bam) }
def tumor_bams = B.out.tumor_output.map { pair_id, bam, bai -> tuple(pair_id, bam) }

// 自动配对同一样本的两类BAM文件
def paired_bams = normal_bams.join(tumor_bams)

process M {
  publishDir params.trim, mode: 'copy'

  input:
  tuple val(pair_id), file(normal_bam), file(tumor_bam)

  output:
  tuple val(pair_id), file(flr2_file), emit: flr
  tuple val(pair_id), file(pon_file), emit: pons

  script:
  flr2_file = "${pair_id}_f1r2.tar.gz"
  pon_file = "${pair_id}_somatic_variants_PON.vcf"
  def normalRGSM = "NORMAL_${pair_id}"

  """
  ./gatk --java-options "-Xmx24G" Mutect2 \
    --native-pair-hmm-threads 20 \
    -R ${params.reference} \
    -I ${tumor_bam} \
    -I ${normal_bam} \
    --normal-sample ${normalRGSM} \
    -pon ${params.ponsfile} \
    -germline-resource ${params.germlineresource} \
    -L ${params.intervalfile} \
    --f1r2-tar-gz ${flr2_file} \
    --output ${pon_file}
  """
}

// 流程启动逻辑
workflow {
  // 替换为实际输入通道，例如从文件对生成输入
  // input_ch = Channel.fromFilePairs("path/to/*_recal.bam")
  A(input_ch)
  B(input_ch)
  M(paired_bams)
}

关键说明 #

• join操作符会自动按tuple的第一个元素（pair_id）匹配通道条目，确保同一样本的文件被正确配对。
• 若需同时传递BAM索引文件，只需调整map和进程输入部分，保留bai字段即可。
• 修正了原脚本中的语法错误，补充了Mutect2必需的-I参数，确保命令格式合规。

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