Nextflow中用emit传递输出至后续步骤及报错解决咨询

问题背景 #

现有Nextflow脚本step1.nf（代码如下）可解析指定注释文件，在results目录生成sample.csv。需将该sample.csv与原始注释文件共同作为sample_parse.py的输入完成后续分析，合并为单个脚本时出现输入通道不匹配的报错，同时需明确后续.id后缀文件的存储与传递方式。

原step1.nf代码：

nextflow.enable.dsl = 2
// Define the default annotation file
params.annotation_file = 'sample_annot.txt'

annotation_ch = Channel.fromPath(params.annotation_file)

// Define the process to parse the annotation file
process parse_annotation {
    tag "$annotation_file"
    publishDir "results/", mode: 'copy'

    input:
    path annotation_file

    output:
    path "sample.csv", emit: parsed_annotation

    script:
    """
    python3 /home/punit/temp/parse_annotation.py -f $annotation_file
    """
}

// Define the workflow
workflow {
    parse_annotation(annotation_ch)
}

更新需求：sample_parse.py预期输出如下，后续需使用其中.id后缀文件调用工具下载数据：

GSM1871919.id  GSM1871921.id  GSM1871923.id  GSM1871925.id  GSM1871927.id  GSM1871920.id  GSM1871922.id  GSM1871924.id  GSM1871926.id  GSM1871928.id 
PrefrontalCortexADEmoryADRC_VS_PrefrontalCortexControlEmoryADRC_C2.info.txt
PrefrontalCortexADEmoryADRC_VS_PrefrontalCortexControlEmoryADRC_C2.samplesheet.csv
PrefrontalCortexADKentuckyADRC_VS_PrefrontalCortexControlKentuckyADRC_C1.info.txt
PrefrontalCortexADKentuckyADRC_VS_PrefrontalCortexControlKentuckyADRC_C1.samplesheet.csv

报错信息：

nextflow run main_v2.nf
N E X T F L O W  ~  version 23.10.1
Launching `main_v2.nf` [hopeful_neumann] DSL2 - revision: d3a1cad6ba
Process `sample_annotation` declares 2 input channels but 1 were specified

 -- Check script 'main_v2.nf' at line: 45 or see '.nextflow.log' file for more details
[ERROR] Terminal initialization failed; falling back to unsupported
java.lang.IllegalStateException: Shutdown in progress
        at java.base/java.lang.ApplicationShutdownHooks.add(ApplicationShutdownHooks.java:66)
        at java.base/java.lang.Runtime.addShutdownHook(Runtime.java:216)
        at jline.internal.ShutdownHooks.addHook(ShutdownHooks.java:79)
        at jline.internal.ShutdownHooks.add(ShutdownHooks.java:43)
        at jline.TerminalSupport.init(TerminalSupport.java:46)
        at jline.UnixTerminal.init(UnixTerminal.java:47)
        at jline.TerminalFactory.create(TerminalFactory.java:101)
        at jline.TerminalFactory.get(TerminalFactory.java:159)
        at nextflow.trace.AnsiLogObserver.renderProcesses(AnsiLogObserver.groovy:249)
        at nextflow.trace.AnsiLogObserver.renderProgress(AnsiLogObserver.groovy:289)
        at nextflow.trace.AnsiLogObserver.render0(AnsiLogObserver.groovy:177)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:568)
        at org.codehaus.groovy.reflection.CachedMethod.invoke(CachedMethod.java:107)
        at groovy.lang.MetaMethod.doMethodInvoke(MetaMethod.java:323)
        at groovy.lang.MetaClassImpl.invokeMethod(MetaClassImpl.java:1254)
        at groovy.lang.MetaClassImpl.invokeMethodClosure(MetaClassImpl.java:1042)
        at groovy.lang.MetaClassImpl.invokeMethod(MetaClassImpl.java:1128)
        at groovy.lang.MetaClassImpl.invokeMethod(MetaClassImpl.java:1030)
        at groovy.lang.Closure.call(Closure.java:427)
        at groovy.lang.Closure.call(Closure.java:406)
        at groovy.lang.Closure.run(Closure.java:498)
        at java.base/java.lang.Thread.run(Thread.java:833)

[-        ] process > parse_annotation -

解决方案 #

报错原因 #

报错核心是新增的后续分析流程声明了2个输入通道，但调用时仅传入1个，导致输入不匹配。需确保流程调用时传递正确数量的输入通道。

完整合并脚本 #

nextflow.enable.dsl = 2

// 参数定义
params.annotation_file = 'sample_annot.txt'
params.parse_script = '/home/punit/temp/parse_annotation.py'
params.sample_parse_script = '/path/to/sample_parse.py'  // 替换为实际路径
params.outdir = 'results'

// 原始注释文件通道
annotation_ch = Channel.fromPath(params.annotation_file)

// 第一步：解析注释文件生成sample.csv
process parse_annotation {
    tag "$annotation_file"
    publishDir "${params.outdir}/parse_annotation", mode: 'copy'

    input:
    path annotation_file

    output:
    path "sample.csv", emit: parsed_csv

    script:
    """
    python3 ${params.parse_script} -f ${annotation_file}
    """
}

// 第二步：使用原始注释文件和sample.csv执行后续分析
process run_sample_parse {
    tag "${annotation_file} + ${sample_csv}"
    publishDir "${params.outdir}/sample_parse", mode: 'copy'

    input:
    path annotation_file  // 原始注释文件
    path sample_csv       // 第一步生成的sample.csv

    output:
    path "*.id", emit: id_files          // 所有.id文件，供后续下载使用
    path "*.info.txt", emit: info_files  // 信息文件
    path "*.samplesheet.csv", emit: samplesheet_files  // 样本表文件

    script:
    """
    python3 ${params.sample_parse_script} --annotation ${annotation_file} --sample-csv ${sample_csv}
    """
}

// 工作流串联
workflow {
    // 执行第一步
    parsed_results = parse_annotation(annotation_ch)
    // 执行第二步：传递原始注释文件和第一步的输出
    run_sample_parse(annotation_ch, parsed_results.parsed_csv)

    // 可选：打印.id文件路径，验证输出
    run_sample_parse.id_files.view()
}

关键说明 #

1. 输入通道匹配：在workflow中调用run_sample_parse时，需同时传入原始注释文件通道annotation_ch和第一步输出的parsed_csv通道，确保与process的两个输入项对应。
2. 输出拆分：通过emit将不同类型的输出拆分为独立通道，后续流程可直接调用run_sample_parse.id_files获取所有.id文件。
3. 目录结构：使用分层的publishDir将不同步骤的结果分开存储，避免文件混乱。
4. 参数化配置：将脚本路径、输出目录等设为参数，提高脚本复用性。

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