DSL2 语法基础：workflow、process、channel 的定义与使用

DSL2 是 Nextflow 在 20.x 版本之后主推的语法体系。说白了，就是把之前 DSL1 那种「脚本里混着流程定义」的写法，拆成了三块清晰的积木：workflow、process、channel。我当初从 DSL1 切到 DSL2 时，整整花了两周才把那些隐式的数据流传透——但一旦搞明白，你就再也不想回到老版本了。

1. 三个核心概念

我们先看看这三块积木分别干什么活：

概念 作用 类比（日常生活）
Channel 数据管道，一个接一个地传递文件或值 生产线上的传送带
Process 执行具体任务（运行命令、脚本、容器） 生产线上的工位
Workflow 编排流程，定义输入->process->输出的顺序和分支 工厂的布局图 + 调度控制室

这三个东西组合在一起，就构成了你整个 WGS 变异检测流程的骨架。嗯，我个人习惯把 channel 比作「水管」，process 是「水龙头」，workflow 就是「管道图纸」。你想想看，没有图纸乱接水管，轻则漏水（数据丢失），重则爆炸（流程卡死）。

2. Channel 的定义与使用

Channel 是 Nextflow 所有数据的入口。没有 channel，process 就拿不到输入文件。

2.1 创建 Channel

最常用的几个 Channel 工厂方法：

// 从文件路径创建
ch_reads = Channel.fromPath("data/sample_*.fastq.gz")
// 从值列表创建
ch_samples = Channel.of("NA12878", "NA12891", "NA12892")
// 从文件中的每行文本创建
ch_ids = Channel.fromPath("samples.txt").splitText()

// 常用的一对一映射
ch_pairs = Channel.fromFilePairs("data/*_{R1,R2}.fastq.gz")

我在项目中踩过一个坑：fromPath 默认使用全局路径匹配，如果你的工作目录不对，它可能匹配到空集合。曾经有一次我 debug 了一下午，才发现是 params.input 没传对路径。后来我在每个流程开头都加一句 .view() 来确认数据是否进来。

2.2 Channel 的操作符

DSL2 里 channel 支持几十种操作符，但刚上手你只需要记住这几个：

  • .view() – 调试专用，打印 channel 里的元素
  • .map{} – 对每个元素做变换
  • .filter{} – 过滤元素
  • .combine() – 两个 channel 做笛卡尔积
  • .join() – 像 SQL 的 inner join

举个实际例子：

ch_reads = Channel.fromFilePairs("data/*_{R1,R2}.fastq.gz")
ch_ids  = Channel.of("NA12878", "NA12891", NA12892")

// 把 read 对和样本 ID 组合成每个样本一个元组
ch_paired = ch_reads
    .map{ sample_id, reads -> 
        [sample_id, reads[0], reads[1]] 
    }
    .join(ch_ids.map{ id -> [id, id] })

注意 .join() 要求两个 channel 的元素类型必须匹配——否则你会得到空的输出。我当初没仔细看文档，以为它会自动做模糊匹配……嗯，结果那个流程跑了 3 小时，全部输出都是空文件。

调试技巧：在 process 前面加一个 ch_input.view()，就能看到 process 会收到什么数据。这是排查「数据走丢了」最直接的方法。

3. Process 的定义与使用

Process 是真正干活的单元。每个 process 都有自己的输入、输出、脚本块和资源定义。

3.1 Process 的基本结构

process FASTQC {
    tag "样本：${sample_id}"
    publishDir "${params.outdir}/qc", mode: 'copy', overwrite: true

    input:
    tuple val(sample_id), path(reads)

    output:
    path "${sample_id}_fastqc.html", emit: html
    path "${sample_id}_fastqc.zip",  emit: zip

    script:
    """
    fastqc -o . -t ${task.cpus} ${reads}
    """
}

关键点：

  • input: 里声明输入的类型：val 表示值类型，path 表示文件路径，tuple 把多个字段打包成一个元组
  • output: 用 emit: 给输出取一个名字，后面 workflow 可以通过名字引用
  • publishDir 控制输出文件的存放位置和方式
  • tag 让日志更容易看懂——千万别省这行，我见过整个流程 200 个 task，日志里全是 "process FASTQC" 根本分不清谁是谁

3.2 Process 的资源声明

DSL2 推荐把资源写在 process 内部或者 profile 里，不要硬编码：

process BWA_MEM {
    cpus 8
    memory 16.GB
    time 2.h

    input:
    tuple val(sample_id), path(fastq1), path(fastq2)

    output:
    tuple val(sample_id), path("${sample_id}.sorted.bam")

    script:
    """
    bwa mem -t ${task.cpus} -R '@RG\\tID:${sample_id}\\tSM:${sample_id}' \
        ${params.reference} ${fastq1} ${fastq2} | \
        samtools sort -@ ${task.cpus} -o ${sample_id}.sorted.bam -
    """
}

注意 ${task.cpus} 是 Nextflow 内置变量，会自动使用你声明的 cpus 数量。如果你在脚本里硬写了 -t 8，但实际只分配了 2 个 CPU——资源浪费是小事，有些工具会报错。

3.3 Process 的模块化

DSL2 最大的改进之一：可以把 process 单独放在 .nf 文件里，然后在主流程中用 include 引入。我个人习惯把每个工具（bwa、gatk、samtools）各写一个 process，放在 modules/ 目录下。

// modules/fastqc.nf
process FASTQC {
    // ... 同上
}

// main.nf
include { FASTQC } from './modules/fastqc'

workflow {
    FASTQC(ch_reads)
}

这样维护起来特别清爽——你改一个工具的版本或参数，只需要改那个模块文件，不会牵扯到其他部分。

4. Workflow 的定义与使用

Workflow 是 DSL2 最核心的部分，也是 DSL1 里没有的东西。它负责把 channel 和 process 串起来。

4.1 定义主 workflow

workflow {
    // 创建输入 channel
    ch_reads = Channel.fromFilePairs("${params.input}/*_{R1,R2}.fastq.gz")
                  .map{ id, reads -> tuple(id, reads[0], reads[1]) }

    // 运行 fastqc
    FASTQC(ch_reads)
    
    // 从 FASTQC 的输出中获取 html 文件，传给下一个 process
    ch_trimmed = TRIMGALORE(ch_reads, FASTQC.out.html)
    
    // 比对
    ch_bam = BWA_MEM(ch_trimmed)
    
    // 跑 GATK 流程
    ch_gvcf = GATK_HAPLOTYPECALLER(ch_bam)
    
    // 收集所有 gvcf
    ch_gvcf.collect()
        .set{ ch_all_gvcf }
    
    // 联合基因分型
    GATK_COMBINEGVCFS(ch_all_gvcf)
}

你可以看到，数据流就像拼乐高一样：前一个 process 的 .out 直接作为下一个的输入。注意 TRIMGALORE 接收了两个输入：reads 和 fastqc 的 html（实际上这里只是为了演示多输入，生产环境一般不需要这样传）。

4.2 带参数的 workflow

你也可以定义带参数的 workflow，方便测试和复用：

workflow {
    take:
    reads_ch

    main:
    ch_qc = FASTQC(reads_ch)
    ch_trim = TRIMGALORE(reads_ch)
    ch_bam = BWA_MEM(ch_trim)
    GATK_HAPLOTYPECALLER(ch_bam)
}

然后在主 workflow 里调用：

workflow {
    ch_all = Channel.fromFilePairs(...)
    MY_WGS_PIPELINE(ch_all)
}

如果你写过其他编程语言，这其实就跟函数调用一个道理。我当初从 DSL1 迁到 DSL2，最不习惯的就是这种显式的数据流——但习惯了之后，你会发现 bug 率直线下降，因为数据从哪来到哪去一目了然。

5. 三者之间的关系图

下面这张图展示了 Channel、Process、Workflow 如何协作完成一个 WGS 流程的典型片段：

DSL2 三要素协作示意（WGS 流程片段） Channel 层：数据源头与变换 fromFilePairs (fastq 文件对) .map { } (提取样本ID) .view() 调试 (打印确认) Process 层：任务执行 FASTQC 输入: tuple(sample_id, reads) 输出: html, zip BWA_MEM 输入: tuple(sample_id, trim_fq) 输出: sorted.bam GATK_HC 输入: sorted.bam 输出: g.vcf.gz Workflow 层：编排与数据流控制 自动追踪依赖、分叉与合并 (无需手动写循环/条件) 输出收集（collect、join、mix） (为联合分析准备数据)

图里把三个层分得很清楚：最上面是 Channel 层，你在这个阶段准备好数据，做基本的清理和检验；中间是 Process 层，每个 process 独立运行，输入输出都有明确的类型；最下面是 Workflow 层，它不执行任何计算任务，只负责把 process 按顺序串联，并处理分支、合并、收集等高级逻辑。

6. 常见避坑指南

坑1：Channel 的数据类型一定要匹配
比如 process 的 input 声明了 tuple val(x), path(y)，但你的 channel 传进来的是 tuple path(x), path(y) 或者顺序反了——Nextflow 不会给你报编译错误，只会静默地输出空结果或报奇怪的运行时错。我吃过这个亏，后来每次写新 process 都先写一个小规模的 .view() 调试一遍。
坑2：不要忘记 emit: 命名
在 DSL2 里，如果你不写 emit: xxx，process 的输出只能通过 .out 的默认索引访问，代码可读性差不说，一旦你改了 process 的输出顺序，调用方直接炸。我个人习惯所有 process 都显式命名每一个输出。
最佳实践：
  • 每个 process 加 tag，便于日志排查
  • 用 publishDir mode: 'copy' 而不是 move，避免调试时重跑导致源文件丢失
  • 在 workflow 开头加一个 ch_input.view()，确认数据加载正确后再跑下去
  • 把 process 拆成独立模块文件，不要在 main.nf 里堆积上百行代码

我刚开始写 DSL2 流程时，因为搞混了 val 和 path，一个 48 样本的 WGS 流程跑了整整一夜，结果发现 BAM 文件全是空的——原因是 fromFilePairs 返回的 path 是文件路径字符串，但我的 process 把它当成了文件对象，导致 ${reads} 展开后是 /path/to/file 这个字符串而不是实际内容。嗯，后来我加了一行 view() 就发现了。

好了，以上就是 DSL2 三个核心积木的定义与用法。你把这部分基础打牢，后面的章节会变得非常顺手——因为整个 WGS 流程本质上就是把这些积木按照变异检测的生物学逻辑拼接起来。记住：先管好数据通道（channel），再定义工位（process），最后用 workflow 画流水线布局图。顺序别搞反了。

← 上一章 📖 返回目录 下一章 →