Nextflow Pipeline异常求助：Python脚本未生成输出文件

问题描述 #

我希望构建一个Nextflow Pipeline，从名为“input”的文件夹读取输入文件，文件格式为Gene_KOs1.tsv、Gene_KOs2.tsv、Mutations1.tsv、Mutations2.tsv这类。我的Python脚本会处理每一对匹配的文件并生成输出，希望输出文件存入如Gene_Mutation1这类命名的子文件夹（因每个输出文件名相同）。但当前编写的代码无法正确保存输出CSV文件，目录为空，无任何Python脚本执行的痕迹。

原代码如下：

#!/usr/bin/env nextflow nextflow.enable.dsl=2

params.inputDir = "/mnt/scratch-raid/data/l/nextflow_test/input" 
params.outputDir = "/mnt/scratch-raid/data/l/nextflow_test/output"


workflow {
    inputDir = file(params.inputDir)
    
    // List all gene files
    genesFiles = inputDir.listFiles().findAll { it.name.startsWith('Gene_KOs') }
    
    // List all mutation files
    mutationFiles = inputDir.listFiles().findAll { it.name.startsWith('Mutations') }
    
    // Process each gene file
    genesFiles.each { genesFile ->
        // Extract the number from the gene file name
        def geneNumber = genesFile.name.replaceAll("[^0-9]", "")

        // Find the matching mutation file
        def mutationFile = mutationFiles.find { it.name.contains("Mutations${geneNumber}") }

        if (mutationFile) {
            println "Starting analysis for ${genesFile.name}"

            genesBaseName = genesFile.name.replaceAll("\\.tsv$", "")
            mutationsBaseName = mutationFile.name.replaceAll("\\.tsv$", "")
            pairOutputDir = "${params.outputDir}/${genesBaseName}_${mutationsBaseName}"

            // Create the output directory
            file(pairOutputDir).mkdirs()

            // Run your Python script here
            """
            python3 /mnt/scratch-raid/data/l/nextflow_test/Main.py ${genesFile} ${mutationFile} ${pairOutputDir}
            """
            
            println "Finished processing: ${genesFile.name} and ${mutationFile.name}"
        } else {
            println "No matching mutations file found for: ${genesFile.name}"
        }
    }
}

问题根源 #

• 命令执行方式错误：在Nextflow DSL2的workflow块中，直接写字符串形式的命令不会被执行。Nextflow要求必须通过process块定义可执行的任务，再通过通道将数据传递给process。
• 违背Nextflow工作流模型：直接在workflow中创建目录、遍历文件的方式不符合Nextflow的设计理念，Nextflow依赖通道（Channel）来管理数据流转，而非传统的文件遍历。
• 输出管理不当：手动创建输出目录并指定路径，会绕过Nextflow的工作目录（work dir）机制，导致输出无法被正确追踪和发布。

修复后的代码 #

#!/usr/bin/env nextflow
nextflow.enable.dsl=2

params.inputDir = "/mnt/scratch-raid/data/l/nextflow_test/input"
params.outputDir = "/mnt/scratch-raid/data/l/nextflow_test/output"

// 定义处理配对文件的process
process runAnalysis {
    tag "${sample_id}"  // 标记任务，方便日志查看
    publishDir "${params.outputDir}/Gene_Mutation${sample_id}", mode: 'copy'  // 自动发布输出到指定子目录

    input:
    tuple val(sample_id), path(gene_file), path(mutation_file)

    output:
    path "*.csv"  // 捕获Python脚本生成的所有CSV文件

    script:
    """
    python3 /mnt/scratch-raid/data/l/nextflow_test/Main.py ${gene_file} ${mutation_file} ./
    """
}

workflow {
    // 使用fromFilePairs自动配对文件，根据文件名中的数字分组
    paired_files = Channel.fromFilePairs("${params.inputDir}/{Gene_KOs,Mutations}{[0-9]*}.tsv", flat: true)
        .map { id, files ->
            // 提取数字编号作为sample_id
            def sample_id = id.replaceAll("[^0-9]", "")
            // 区分基因文件和突变文件
            def gene_file = files.find { it.name.startsWith('Gene_KOs') }
            def mutation_file = files.find { it.name.startsWith('Mutations') }
            tuple(sample_id, gene_file, mutation_file)
        }

    // 将配对好的数据传入process执行
    runAnalysis(paired_files)
}

关键改进点 #

• 使用fromFilePairs配对文件：自动根据文件名中的模式分组配对，避免手动遍历和匹配的错误。
• 用process定义任务：符合Nextflow的执行模型，确保命令被正确调度执行。
• publishDir自动管理输出：将process生成的CSV文件自动复制到指定的子目录（如Gene_Mutation1），无需手动创建目录。
• 通道传递数据：通过通道流转文件和元数据，保证工作流的可重复性和可追踪性。

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