4. 流程核心概念：Channel、Process、Workflow、Operator

说实话，刚接触 Nextflow 的时候，我花了一周才彻底搞明白这四个东西的关系。你给我讲讲你的理解——是不是觉得 Channel 像水管，Process 像加工站，Workflow 像车间布局图？这么理解其实没错，但真要上手写流程，细节里全是坑。今天我就把我踩过的坑和摸清的门道都摆出来说。

4.1 Channel：数据流动的血管

Channel 是 Nextflow 里一切数据的来源。说白了，它就是一条单向的数据管道。数据从源头进去，经过各种变换，最后被 Process 消费掉。

我一开始犯的错误——以为 Channel 只是个列表。后来才发现，Channel 是异步的、流式的。这俩词啥意思？举个例子：你有一个文件列表要处理，普通循环是一次处理一个，但 Channel 可以边产生数据边消费，处理完的文件不用等后面的。

创建 Channel 最常见的方式就三种：

// 从一个值创建
ch_string = Channel.of('hello', 'world', 'nextflow')

// 从文件路径创建
ch_fastq = Channel.fromPath('/data/samples/*.fastq.gz')

// 从列表创建
ch_samples = Channel.fromList(['S1', 'S2', 'S3'])

这里有个关键点：Channel 分为两种，queue channel 和 value channel。Queue channel 像传送带，数据流过去就没了；value channel 像个固定招牌，你随时可以读它，读一万次内容都一样。

类型 创建方式 特性 我踩过的坑
queue channel Channel.of、Channel.fromPath 一次性消费，用完即止 同一个 channel 传给两个 process，第二个 process 收不到数据
value channel Channel.value 可重复读，类似常量 忘了用 value channel 传参数，导致参数被 process 消费掉
⚠️ queue channel 只能被消费一次。我在项目里把同一个 channel 同时传给对齐和质控两个 process，结果质控那个 process 什么数据都没收到。排查了半天才发现是 channel 被前一个 process 消耗光了。

要避免这个问题，要么用 .into() 把 channel 复制成多份，要么把全局参数用 value channel 传递。

// 正确做法：用 into 拆分
ch_reads = Channel.fromPath('/data/*.fastq')
ch_reads
  .into { ch_align; ch_qc }

// 参数用 value channel
params.genome = 'hg38'
ch_genome = Channel.value(params.genome)

4.2 Process：核心工作单元

Process 是所有计算发生的场所。每个 process 有三个核心部分：输入声明、输出声明、执行脚本。你想想看，这三者必须严格对应，少了哪个 process 都跑不起来。

我刚开始写 process 最简单的方式是这样：

process ALIGNMENT {
  input:
  path fastq

  output:
  path "*.bam"

  script:
  """
  bwa mem -t 4 $fastq | samtools sort -o ${fastq.baseName}.bam
  """
}

看着很简单对吧？但实际项目中，process 的 directive（指令） 才是真正拉开档次的地方。指令决定了 process 跑在什么资源上、失败怎么重试、要不要用容器。

我的习惯：每个 process 开头至少写 cpus、memory、container 三个指令。哪怕默认值够用，也要显式写出来，以后调参时一眼就能看出每个步骤的资源配置。

process ALIGNMENT {
  cpus 4
  memory '8 GB'
  container 'biocontainers/bwa:v0.7.17_cv1'
  tag "${fastq}"

  input:
  path fastq

  output:
  path "*.bam", emit: bam

  script:
  """
  bwa mem -t ${task.cpus} $fastq | samtools sort -o ${fastq.baseName}.bam
  """
}

你注意到了没有？${task.cpus} 这个变量直接引用了前面定义的 4 核。这样资源变更时，脚本命令也会同步更新，不会出现「配了 8 核但脚本里还在用 4 核」的尴尬。

💡 tag 指令是我极力推荐的。它让 nextflow log 里每条记录都标上样本名，排查错误时一眼就能定位到是哪个样本挂了。没有 tag 的话，日志里全是 process 编号，根本对不上样本。

4.3 Workflow：编排的指挥官

Workflow 是 Nextflow 2.x 引入的新语法。它把整个流程的编排逻辑集中到一块。我当初从旧语法迁移过来时，最大的感受是——代码结构清晰了十倍。

workflow {
  // 定义数据源
  read_pairs_ch = Channel.fromFilePairs('data/*_{R1,R2}.fastq.gz')

  // 执行流程
  FASTQC(read_pairs_ch)
  TRIMMOMATIC(read_pairs_ch)
  ALIGNMENT(TRIMMOMATIC.out.trimmed_reads)
  QUANTIFICATION(ALIGNMENT.out.bam)

  // 收集结果
  QUANTIFICATION.out.view()
}

这是完整的工作流描述。你看，从上到下读一遍，就能知道整个流程干了什么：先质控、再修剪、再比对、最后定量。每个 process 的输入输出通过 channel 的 .out 属性串联起来。

我曾经在一个基因组分析项目里，把流程拆成了 12 个 process。如果用旧语法，主脚本得写 200 行，到处是 from 和 into。用 workflow 重构后，主块的代码不到 30 行，每个 process 的依赖关系一目了然。

4.4 Operator：数据的魔法师

Operator 是对 channel 里的数据进行变换的函数。你不会用到全部的 50 多个 operator，但几个核心的必须掌握。

我按实用频率排个序：

  1. map：给每个元素做变换，最常用
  2. filter：按条件筛选数据
  3. flatten：把嵌套列表摊平
  4. groupTuple：按某个字段分组
  5. branch：按条件分流到不同分支
  6. combine：两个 channel 做笛卡尔积
  7. join：按 key 匹配合并 channel
// map 示例：给样本名加前缀
Channel.of('S1', 'S2', 'S3')
  .map { it -> "sample_${it}" }
  .view()
// 输出：sample_S1, sample_S2, sample_S3

// filter 示例：只保留 R1 文件
Channel.fromPath('/data/*.fastq')
  .filter { it.name.contains('R1') }
  .view()

// groupTuple 示例：配对 R1 和 R2
Channel.fromFilePairs('data/*_{R1,R2}.fastq.gz')
  .groupTuple()
  .view()

为什么要强调 groupTuple？因为实际项目中，R1 和 R2 文件往往是通过 glob 模式配对到的双端测序文件。不分组的话，每个文件单独进 process，但比对工具需要同时输入一对。groupTuple 正好能解决这个痛点。

一个实战技巧：用 .combine() 配合 value channel 来传全局参考文件。比如比对时每个样本都需要参考基因组，但参考文件本身不变。用 combine 可以“粘”到每个样本的数据上：

samples_ch = Channel.fromPath('/data/*.fastq')
ref_ch = Channel.value('/ref/hg38.fa')

samples_ch
  .combine(ref_ch)
  .view()
// 输出：[/data/S1.fastq, /ref/hg38.fa]
//       [/data/S2.fastq, /ref/hg38.fa]

4.5 四者关系一览图

⚠️ 上面提到过同一个 channel 不能传递给两个 process，这里结合 operator 的 .into 再强调一次：拆分成多个拷贝后，每个拷贝独立消费，不会被干扰。

ch_fastq
  .into { ch_qc; ch_align }
// ch_qc 和 ch_align 是两个独立的 channel，内容相同，互不影响

最后，我建议把所有 operator 链式调用的逻辑都写在 workflow 块里，不要分散到各个 process 中。这样改流程时，你只用改 workflow 块里的 operator 链，不用翻到每个 process 内部去调。这是我能给你最实用的一个建议了。

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