Nextflow 基础语法与流程定义：DSL2 结构、main.nf 与 config 文件、通道与进程

说实话，刚开始接触 Nextflow 的时候，我被 DSL2 的文件结构绕晕过。后来我仔细理了理，发现其实就三个概念：模块、主流程、配置。说白了，就是把一个复杂的数据分析流程拆成可复用的积木，再拼起来跑。

DSL2 结构：模块化是你的好帮手

Nextflow 的 DSL2（Domain Specific Language 第二版）是 2020 年左右推出来的。我个人习惯，只要项目超过 3 个进程，就立刻用 DSL2。为什么？因为它强制你写 module 文件，进程、通道定义都放在独立文件里，主流程只负责串联。

一个典型的 DSL2 项目结构长这样：

workflow/
  ├── main.nf
  ├── nextflow.config
  └── modules/
      ├── align.nf
      ├── quantify.nf
      └── stats.nf

主文件 main.nf 只做两件事：引入模块 和 定义工作流。模块文件里写具体的 process 和 channel。我最早踩过的坑，就是把所有进程堆在一个文件里，后来维护时差点崩溃。

main.nf：流水线的指挥棒

main.nf 是 Nextflow 的入口。DSL2 里，你需要用 include 语句把模块里的进程导入进来。有点像 Python 里的 from xxx import yyy。

来看个最简单的示例：

nextflow.enable.dsl=2

include { SAMTOOLS_SORT } from './modules/samtools'
include { BWA_MEM } from './modules/bwa'

workflow {
    BWA_MEM()
    SAMTOOLS_SORT(BWA_MEM.out.bam)
}

注意 nextflow.enable.dsl=2 这个声明，必须写在最前面。我曾经忘记写这一行，结果报了一堆莫名其妙的错，检查了半天才发现。

nextflow.config：不要硬编码

nextflow.config 文件用来统一管理参数。比如参考基因组路径、样本列表、资源配额等。我一般把配置文件写在项目根目录，然后所有模块里的 params 变量都从 nextflow.config 里读。

看个配置示例：

params {
    genome = '/data/hg38.fa'
    samples = '/data/samples.csv'
    outdir = './results'
    max_cpus = 4
    max_memory = '16.GB'
}

process {
    executor = 'local'
    cpus = params.max_cpus
    memory = params.max_memory
}

为什么推荐这样做？你想想看，如果不把参数抽离出来，每次换一个数据集就要手动改 main.nf，容易改出 bug。我曾在生产环境因为硬编码路径错误，重跑了一整天的任务，血的教训。

通道（Channel）：数据流动的生命线

通道是 Nextflow 里最核心的概念，没有之一。它负责把数据从一个进程送到另一个进程。你可以把通道想象成一根水管，数据就像水流，水管可以分叉、合并、过滤。

常见的创建方式：

// 从文件列表创建
Channel.fromPath('*.fastq.gz')

// 从值创建
Channel.value('hg38')

// 从 CSV 文件解析
Channel.fromPath(params.samples).splitCsv(header: true)

我最常用的是 .splitCsv 和 .map。举个例子，我的样本信息表长这样：

sample_id,read1,read2
S1,/data/S1_R1.fq,/data/S1_R2.fq
S2,/data/S2_R1.fq,/data/S2_R2.fq

然后我可以用通道把它变成 [sample_id, reads] 的元组，方便下游并行处理：

samples_ch = Channel
    .fromPath(params.samples)
    .splitCsv(header: true)
    .map { row -> [row.sample_id, file(row.read1), file(row.read2)] }

为什么要关注通道的类型？因为进程的 input 和 output 定义决定了通道里流动的数据结构。我见过有人把 val 和 file 搞混，结果进程接收到了文件对象却当字符串处理，当然会报错。

进程（Process）：干活的小兵

每个进程代表一个独立的任务单元。它包含 input、output、script 三个主要部分，以及可选的 publishDir、executor 等。

一个标准的进程定义：

process BWA_MEM {
    tag { "${sample_id}" }
    publishDir "${params.outdir}/bam", mode: 'copy'

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

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

    script:
    """
    bwa mem -t $task.cpus /data/hg38.fa $reads > ${sample_id}.sai
    bwa samse /data/hg38.fa ${sample_id}.sai $reads > ${sample_id}.bam
    """
}

注意几点：

• tag 帮你在运行日志里快速定位任务，我每次必加。
• publishDir 把结果复制到输出目录，但模式建议用 copy 而不是 move，避免影响缓存。
• $task.cpus 会自动获取 config 里设定的 CPU 数。我刚开始总写成 -t 4，hard code 了数字，后来改成 $task.cpus 后资源调度灵活多了。

流水线执行流程：一张图理清

看完概念，我们来把它们串起来。下面这张图展示了 Nextflow 从输入到输出的完整流动过程：

流程其实很简单：

1. 输入通道 读取数据（比如 fastq 文件），自动分发到进程 A。
2. 进程 A 执行比对，产生 bam 文件，通过输出通道传递给进程 B。
3. 进程 B 执行排序，输出最终结果到 输出通道，同时 配置文件 为所有进程提供全局资源设置。
4. 整个流程的并行度由 Nextflow 根据有效资源自动调度，你只需关注逻辑关系。

避坑指南：我亲自踩过的雷

• DSL2 未启用：在 main.nf 最顶部写 nextflow.enable.dsl=2，否则就无法使用 include 和模块化。
• 通道数据格式不匹配：比如进程输入声明了 tuple val(sample_id), path(bam)，但上游通道给了 [bam, sample_id]，顺序反了会报错。我建议在通道链上多打印几次 .view() 来调试。
• 资源硬编码：永远别在脚本里写 samtools sort -@ 8，改成 $task.cpus，这样切换到集群时就能自动适配。
• publishDir 误用 move：用过 mode: 'move' 后，缓存里的文件就没了，重新运行需要重算。我一般用 mode: 'copy' 或 symlink。

小结