基准通道创建NextFlow

Question

我正在NextFlow上执行一个分散-聚集操作。

它看起来如下所示：

reads = PATH+"test_1.fq"
outdir = "results"

split_read_ch = channel.fromFilePairs(reads, checkIfExists: true, flat:true ).splitFastq( by: 10, file:"test_split" )

process Scatter_fastP {
    tag 'Scatter_fastP'
    publishDir outdir

    input:
    tuple val(name), path(reads) from split_read_ch
    
    output:
    file "${reads}.trimmed.fastq" into gather_fatsp_ch

    script:
    """
    fastp -i ${reads} -o ${reads}.trimmed.fastq
    """
}


gather_fatsp_ch.collectFile().view().println{ it.text }

我使用Nextflow (https://www.nextflow.io/docs/latest/tracing.html)建议的所有基准测试选项运行此代码：nextflow run main.nf -with-report nextflow_report -with-trace nextflow_trace -with-timeline nextflow_timeline -with-dag nextflow_dag.html

在这些跟踪文件中，我可以找到10个Scatter_fastP进程的资源和速度。但我还想测量一下创建split_read_ch和gather_fastp_ch通道的资源和速度。

我试图在流程中包含通道的创建，但我找不到一个解决方案来使其工作。有没有一种方法可以将通道创建包含到跟踪文件中？或者，有没有一种我还没有找到的将这些通道创建到进程中的方法？

提前感谢您的帮助。

Answer 1

虽然Nextflow可以解析FASTQ文件并将它们拆分成更小的文件等，但通常情况下，最好将这些操作传递给另一个或一组进程，特别是如果您的输入FASTQ文件很大的话。这在两个方面是有益的：(1)您的主要下一个流程不需要如此努力地工作，(2)您可以在您的下一个流程报告中获得粒度任务流程统计信息。

下面的示例使用GNU split来拆分输入的FASTQ文件，并使用groupTuple()操作符和groupKey()内建的方法收集输出，以便尽可能快地流式传输收集的值。你需要适应你的非gzipped输入：

nextflow.enable.dsl=2

params.num_lines = 40000
params.suffix_length = 5

process split_fastq {

    input:
    tuple val(name), path(fastq)

    output:
    tuple val(name), path("${name}-${/[0-9]/*params.suffix_length}.fastq.gz")

    shell:
    '''
    zcat "!{fastq}" | split \\
        -a "!{params.suffix_length}" \\
        -d \\
        -l "!{params.num_lines}" \\
        --filter='gzip > ${FILE}.fastq.gz' \\
        - \\
        "!{name}-"
    '''
}

process fastp {

    input:
    tuple val(name), path(fastq)

    output:
    tuple val(name), path("${fastq.getBaseName(2)}.trimmed.fastq.gz")

    """
    fastp -i "${fastq}" -o "${fastq.getBaseName(2)}.trimmed.fastq.gz"
    """
}

workflow {

    Channel.fromFilePairs( './data/*.fastq.gz', size: 1 ) \
        | split_fastq \
        | map { name, fastq -> tuple( groupKey(name, fastq.size()), fastq ) } \
        | transpose() \
        | fastp \
        | groupTuple() \
        | map { key, fastqs -> tuple( key.toString(), fastqs ) } \
        | view()
}