如何让Nextflow流水线按谱系数据集列表长度并行多次运行并生成报告

核心问题梳理 #

当前流水线存在以下关键问题：

• BUSCO流程脚本存在语法错误，导致物种信息无法正确追加到summary文件
• 数据流转逻辑未正确收集所有BUSCO输出文件
• createReport流程输入配置未正确接收所有summary文件并映射到指定目录
• report.py依赖硬编码路径，无法适配Nextflow的工作目录机制

具体修改内容 #

1. 修复BUSCO流程的脚本与输出配置 #

修改main.nf中的busco流程，修正语法错误并确保输出文件能正确传递：

process busco {
  container "quay.io/biocontainers/busco:5.2.2--pyhdfd78af_0"

  tag "${lineage_dataset}: ${input_file.name}"

  // 可选：将中间结果发布到指定目录，方便调试查看
  publishDir params.outdir.intermediate, mode: 'copy', saveAs: "${lineage_dataset}_${input_file.baseName}_short_summary.txt"

  input:
    path input_file
    each lineage_dataset
  
  output:
    path "hap1/short_summary.${lineage_dataset}.${output_name}.txt", emit: busco_summary

  script: 
    output_name = input_file.baseName

    """
    busco \
    -m ${params.mode} \
    -o hap1 \
    -i $input_file \
    -l ${lineage_dataset} \
    --augustus_species ${params.augustus_species} \
    --update-data \
    -c 8

    // 修正echo语法，将物种信息追加到summary文件
    echo "${params.augustus_species}" >> "./hap1/short_summary.${lineage_dataset}.${output_name}.txt"
    """
}

2. 调整createReport流程的输入与脚本 #

修改main.nf中的createReport流程，确保能接收所有BUSCO输出文件并正确调用报告脚本：

process createReport {

  publishDir params.outdir.main, mode: 'copy'

  input: 
    // 接收所有BUSCO输出的summary文件，自动映射到busco_outputs目录
    path busco_summary, stageAs: 'busco_outputs/${fileName}'

  output:
    path 'report.html'

  script:
    """
    // 激活虚拟环境（根据实际路径调整）
    source "${launchDir}/venv/bin/activate"
    // 执行报告生成脚本，将输出写入指定文件
    python report.py > report.html
    """ 
}

3. 修正Workflow的数据流转逻辑 #

修改main.nf中的workflow部分，确保所有BUSCO输出文件被收集并传递给createReport：

workflow {
  input_files = Channel.fromPath(params.input_files)
  
  // 每个输入文件与每个谱系数据集配对，并行执行BUSCO任务
  busco(input_files, params.lineage_datasets)
    .busco_summary // 引用BUSCO流程的指定输出
    .collect() // 收集所有summary文件到一个集合
    .set { all_busco_summaries }
  
  createReport(all_busco_summaries)
}

4. 修复report.py的路径读取逻辑 #

修改report.py，适配Nextflow的工作目录机制，直接读取当前流程的busco_outputs目录：

#!/usr/bin/env python

from report_modules.utils.report_utils import Report_Printer
from report_modules.utils.parsing_utils import Report_Parser
from pathlib import Path

// 直接读取当前目录下的busco_outputs文件夹
path = Path("./busco_outputs")
list_of_files = path.glob('*.txt')

all_stats_dicts = {}
for data in list_of_files:
    with open(data, 'r') as file:
        file_data = file.read()
    // 用文件名作为统计字典的键，更易识别
    parser = Report_Parser(file_data)
    all_stats_dicts[data.name] = parser.parse_report()

if __name__ == '__main__':
    report_printer = Report_Printer()
    report_template = report_printer.print_template(all_stats_dicts)
    // 确保输出到标准输出，供Nextflow捕获生成report.html
    print(report_template)

5. 可选：优化资源配置（Nextflow.config） #

根据BUSCO实际运行需求调整资源参数：

process {
    executor                = 'slurm'
    // 为BUSCO流程单独配置资源（匹配BUSCO的-c 8参数）
    withName: busco {
        cpus                    = 8
        time                    = '2hours'
        memory                  = '16.GB'
    }
    withName: createReport {
        cpus                    = 4
        time                    = '30minutes'
        memory                  = '4.GB'
    }
}

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