如何在Nextflow中实现进程间输出传递及并行处理JSON文件

需求说明 #

已实现create_parallel_params进程，可生成包含0.json至9.json的parallel_params文件夹（由parallel_paramgen.py生成），需将这些JSON参数文件传入searchlibrarysearch_molecularv2_parallelstep1进程，实现多文件并行处理。

修改后的完整Nextflow代码 #

1. 参数生成进程（create_parallel_params） #

process create_parallel_params {
    publishDir "./nf_output", mode: 'copy'
    input:
    val x
    output:
    // 输出文件夹下所有JSON文件，每个文件作为通道独立元素
    path 'parallel_params/*.json', emit: param_jsons

    script:
    """
    mkdir -p parallel_params
    python ${TOOL_FOLDERS}/parallel_paramgen.py \
        parallel_params \
        ${x}
    """
}

2. 并行处理进程（searchlibrarysearch_molecularv2_parallelstep1） #

process searchlibrarysearch_molecularv2_parallelstep1 {
    publishDir "./nf_output", mode: 'copy', pattern: 'intermediateresults_*/*'
    input:
    // 接收单个并行参数JSON文件
    path param_json
    path spectra
    path library
    output:
    // 为每个任务生成独立结果目录，避免写入冲突
    path "intermediateresults_${param_json.baseName}", emit: step1_results

    script:
    """
    mkdir -p intermediateresults_${param_json.baseName} convert_binary librarysearch_binary
    python ${TOOL_FOLDERS}/searchlibrarysearch_molecularv2_parallelstep.py \
        --parallelism 1 \
        ${spectra} \
        ${param_json} \
        ${params.workflow_parameter} \
        ${library} \
        intermediateresults_${param_json.baseName} \
        convert_binary \
        librarysearch_binary
    """
}

3. 工作流编排 #

workflow {
    // 初始化输入值，示例为1
    x = Channel.from(1)
    // 获取生成的所有参数JSON文件通道
    param_jsons_ch = create_parallel_params(x).param_jsons

    // 将每个参数文件与全局的spectra、library文件组合，生成并行任务输入
    step1_input_ch = param_jsons_ch.map { json_file ->
        [json_file, params.spectra, params.library]
    }

    // 启动并行处理，每个JSON文件对应一个独立进程
    searchlibrarysearch_molecularv2_parallelstep1(step1_input_ch)
}

关键修改说明 #

• 参数进程输出调整：将输出从整个文件夹改为单个JSON文件，确保通道中每个元素对应一个待处理参数，为并行提供基础；同时修正目录创建命令（用&&替代|），避免管道执行错误。
• 处理进程输入与输出优化：改为接收单个JSON文件，同时为每个任务生成带参数文件名的独立结果目录，防止多进程同时写入同一目录导致的冲突。
• 工作流并行编排：通过map操作将每个参数文件与全局资源（spectra、library）组合，Nextflow会自动为通道中每个元素启动独立进程，实现多文件并行处理。

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