Nextflow 流程开发

Nextflow Developing Pipelines

nextflow developing piplines中的scripts，processes，workdlows，agents等

前面基础入门已经学习了如何安装nextflow，命令行使用入门，基础的设置文件，执行器，缓存和报告等内容，也能本地运行第一个脚本，这篇主要学习piplines是如何搭建起来的，各个部分该如何了解学习

Nextflow 脚本 = Groovy 编排层 + Bash/Python/R 等执行层

1. Scripts

这一节主要是讲 Nextflow 脚本里 Groovy 语言部分的写法，用来做流程编排

Nextflow是一种运行在Java虚拟机（JVM）上的工作流语言。Nextflow的语法与Groovy非常相似，Groovy是一种用于JVM的脚本语言。但是，Nextflow专门用于以声明式方式编写计算管道。本页是Nextflow语言的实用介绍

1.1 基础变量

章节	说明	示例代码
Hello world	使用 println 函数输出到控制台	println 'Hello, World!'
Variables	使用 def 关键字声明变量，变量可持有任何类型的值，可重新赋值	def num = 1 def str = "Hi" def x = 1 x = 'hello'
Strings	字符串可用单引号或双引号定义；双引号支持插值（$var 和 ${expression}）；三引号可定义多行字符串	println "he said cheese once" def a = "world" println "hello " + a def text = """hello there"""
Lists	使用方括号定义列表，通过方括号索引访问（索引从 0 开始），用 size 方法获取长度	def myList = [1776, -1, 33] println myList[0] println myList.size()
Maps	用于存储键值对（key-value pairs），是无序的命名值集合，可通过 key 或点号访问	def scores = ["Brett": 100] println scores["Pete"] println scores.Pete scores["Pete"] = 3
Records and tuples	Records 存储一组相关字段（可不同类型），通过字段名访问；Tuples 存储固定序列的异构值，通过索引访问，支持解构；两者均不可变	def person = record(name: 'Alice', age: 42) def coords = tuple(1, 2) def (x, y) = coords

1.2 运算符

章节	说明	示例代码
相等/不等	== 和 != 测试两个值是否相等（或不等）	assert 2 + 2 == 4 assert [2, 2] != [4]
比较运算符	比较两个值（数字按数值比较，字符串按字母顺序比较）	assert 3 < 3.14 assert 'hello' < 'world'
逻辑运算符	执行布尔逻辑：AND、OR、NOT	assert (true && false) == false assert (true || false) == true assert (!true) == false
in 运算符	测试成员资格，即集合是否包含某个值	assert 2 in [1, 2, 3]
算术运算符	加减乘除、幂运算（**）、取模（%）；+ 还可拼接列表、Map 和字符串	assert 2 + 2 == 4 assert 2 ** 8 == 256 assert 7 % 2 == 1 assert [1,2,3] + [4] == [1,2,3,4]
assert	测试条件，如果条件为假则抛出错误	assert 2 + 2 == 4

1.3 正则表达式

用法	说明	示例
=~	检查字符串中是否包含某个模式	assert 'hello world' =~ /hello/
==~	检查字符串是否完全匹配某个模式	assert 'hello' ==~ /hello/
replaceFirst	替换第一个匹配项	'colour'.replaceFirst(/ou/, 'o') == 'color'
replaceAll	替换所有匹配项	'cheesecheese'.replaceAll(/cheese/, 'nice') == 'nicenice'
捕获组	用 =~ 返回 matcher 对象，通过索引提取分组	def m = version =~ /(\d+)\.(\d+)\.(\d+)-?(.+)/
m[0]	返回完整匹配和所有捕获组	['2.7.3-beta', '2', '7', '3', 'beta']
m[0][1]	第一个捕获组	'2'
m[0][2]	第二个捕获组	'7'
m[0][3]	第三个捕获组	'3'
m[0][4]	第四个捕获组	'beta'

1.4 闭包

Closure（闭包）可以理解为：把一段代码当作一个可以传递、调用的值，用来告诉程序“拿到每个数据后，应该怎么处理它”

用法	说明	示例
闭包定义	可以像普通值一样使用的函数	{ v -> v * v }
参数与返回	-> 左边是参数，右边是返回值	{ v -> v * v } 表示参数 v，返回 v * v
collect	对列表中每个元素应用闭包，生成新列表	[1,2,3,4].collect { v -> v * v } == [1,4,9,16]
each	遍历集合，对每个元素执行闭包	list.each { item -> println item }
Map 的 each	闭包接收两个参数：key 和 value	map.each { key, value -> println "$key $value" }
闭包用途	常作为高阶函数的参数，以声明式方式处理集合	collect、each、find、filter 等

1.5 Script definitions

前几节介绍了Nextflow代码的基本构建块：变量、字符串、集合和闭包。

在实践中，Nextflow脚本由工作流、过程、函数和类型（统称为定义）组成，并且可以包括来自其他脚本的定义。

要将代码片段转换为工作流脚本，请将其封装在工作流块中：

workflow {
    println 'Hello!'
}

这个块称为输入工作流。它是脚本执行时的入口点。当一个脚本只包含简单的语句，如println ’Hello！， Nextflow将其视为入口工作流

def sayHello() {
    println 'Hello!'
}

def add(a, b) {
    a + b
}

workflow {
    sayHello()
    println "2 + 2 = ${add(2, 2)}!"
}

可以通过将这些定义移动到单独的脚本并将它们包含在主脚本中来进一步组织这些定义

// main.nf
include { sayHello; add } from './functions.nf'

workflow {
    sayHello()
    println "2 + 2 = ${add(2, 2)}!"
}

2. Working with files

常用的文件操作命令

def 是 Groovy 里声明变量的关键字，相当于“定义一个变量”,Groovy 是动态类型语言，def 表示“我不指定具体类型，让运行时自己推断”

def myFile = file('data/sample.txt')   // ✅ 推荐
myFile = file('data/sample.txt')       // ⚠️ 不推荐，变成全局变量

2.1 文件检索（Retrieving files）

Path 对象 = “带智能方法的路径对象”，不是普通字符串，也不是当前目录.

形式	是什么	能做什么
'some/path/file.txt'	普通字符串	只能当文本用
file('some/path/file.txt')	Path 对象	能取文件名、父目录、判断是否存在、读内容

其它用法:

用法	说明	示例
file()	获取单个文件或目录的引用	myFile = file('some/path/to/my_file.file')
files()	获取文件列表，支持通配符 glob 模式	listOfFiles = files('some/path/*.fa')
通配符 * ? [] {}	作为 glob 模式匹配多个文件，无匹配时返回空列表	files('data/*.fastq.gz')
file() 单匹配	file() 也可传 glob，但前提是只匹配一个文件	file('data/sample_01.fa')
`| 双星号不仅匹配当前目录，还会递归搜索子目录 |files(‘some/path//.fa’)| |hidden: true| 默认通配符不匹配隐藏文件，加此选项可包含 |file(’some/path/.fa’, hidden: true)| |resolve()| 基于已有路径引用，拼接相对路径 |dir.resolve(‘sample.bam’)| |/运算符 | 与resolve()等价，写法更简洁 |dir / ‘sample.bam’| |file(“\(dir/...")` | 字符串插值拼接后转回文件引用，可行但冗余 | `file("\)dir/sample.bam”)| |“\(dir/..."` | 纯字符串拼接，不是文件引用，不能当文件用 | `"\)dir/sample.bam”` ❌

file() 函数返回的不是一个简单的字符串，而是一个 Path 对象。这个对象自带了很多属性，可以让你直接读取文件的信息，而不需要自己去解析路径字符串

属性	值	含义
path.name	'file.txt'	完整文件名（含扩展名）
path.baseName	'file'	文件名去掉扩展名的部分
path.extension	'txt'	文件扩展名
path.parent	'/some/path'	文件所在的目录路径

2.2 Reading and writing

读取文件就像使用文件的text属性一样简单，它会以字符串的形式返回文件内容：

def 是 Groovy 里声明变量的关键字，相当于“定义一个变量”

方法	行为	适用场景
myFile.text	读取整个文件，返回字符串	小文件，一次性读取
myFile.text = 'xxx'	覆盖写入；文件不存在会自动创建	小文件，一次性写入
getText() / setText()	与 .text 等价，只是显式方法调用	同上
readLines()	按行读取，返回一个列表	需要把所有行放进内存处理时
eachLine { line -> ... }	逐行加载，每次只把一行放进内存	大文件、日志、FASTA/FASTQ 逐行处理
withReader { r -> ... }	获得 Reader，可逐行或逐字符读取	只需读文件的一部分，比如只读第一行
append('xxx')	在文件末尾追加内容	追加写入
<<	与 append() 等价，写法更短	追加写入

注记

上述方法一次读取和写入整个文件内容，需要将整个文件加载到内存中。考虑为大文件使用更节省内存的方法，例如逐行读取/写入文件

file('some/my_file.txt')
    .readLines()
    .each { line ->
        println line
    }

这个写法等价于

def myFile = file('some/my_file.txt')
def lines = myFile.readLines()
lines.each { line ->
    println line
}

2.3 Filesystem operations

文件操作，主要是用于查看文件夹中的文件内容，包括目录，名称，大小等

方法	用法	说明
list()	file('path').list()	列出目录内容，返回列表
listDirectory()	file('path').listDirectory()	listFiles() 的替代方法（26.04 起推荐）
eachFile {}	myDir.eachFile { item -> ... }	遍历目录每一项，逐个处理
isFile()	item.isFile()	判断是否为文件
isDirectory()	item.isDirectory()	判断是否为目录
getName()	item.getName()	获取文件或目录名称
size()	item.size()	获取文件大小（字节）

2.4 Remote files

Nextflow使用与本地文件相同的接口处理多种类型的远程文件和对象。支持以下协议：

• HTTP(S)/FTP (http://, https://, ftp://)
• Amazon S3 (s3://)
• Azure Blob Storage (az://)
• Google Cloud Storage (gs://)

当一个 process 的输入文件在远程位置，比如对象存储（S3、GCS、Azure Blob）或其他文件系统，而 Nextflow 的工作目录在本地或共享文件系统上时，Nextflow 会先把这些远程文件 复制到工作目录，这个过程叫 remote file staging。

work/stage-<session-id>/<hash>/<filename>

如果大规模运行时，输入文件都在对象存储里，但工作目录在共享文件系统上，Nextflow 要负责把所有文件先搬过来，就会变慢。因为所有文件传输都由 Nextflow 统一处理，容易形成 IO 瓶颈

方式	做法	效果
自定义下载 process	写一个专门下载文件的 process，输入用 val 而不是 path	绕过 Nextflow 内置远程暂存，多个文件可以并行下载
使用 Fusion	把工作目录设在对象存储上，并启用 Fusion	task 直接访问远程文件，不需要预先复制

3. Processes

在 Nextflow 里，process 是一个“专门用来执行脚本”的结构，它能让脚本更容易扩展、更容易在不同环境里运行。

主要做3件事

部分	作用
input	定义这个步骤需要什么输入
output	定义这个步骤会产生什么输出
script	定义实际要执行的命令或脚本

process hello {
    output:
    path 'hello.txt'

    script:
    """
    echo 'Hello world!' > hello.txt
    """
}

3.1 Script

1. 脚本基础写法

这里的脚本更像是process中的流程，而不是第一节的语法。

脚本部分以字符串表达式的形式定义进程执行的脚本。

要点	说明
script 是什么	一个字符串表达式，定义了 process 要执行的脚本
数量限制	一个 process 只能有一个 script 块
位置规则	如果不显式写 script: 关键字，那脚本必须是 process 定义的最后一条语句
执行方式	以 Bash 脚本的形式在宿主环境里执行
可用命令	只能使用宿主环境里实际存在的命令
字符串形式	可以是单行字符串，也可以是多行字符串（推荐多行，方便写多步命令）

推荐用多行字符串

script:
"""
fastqc $reads
samtools sort -o sorted.bam input.bam
samtools index sorted.bam
"""

script 块里如果写了 $XXX，Nextflow 会尝试把它当成 Groovy/Nextflow 变量 做字符串插值。但有时候你想用的是 Bash 环境变量，比如 \(PATH、\)HOME、$DB，这时候就需要区分

场景	写法	示例
不需要 Nextflow 变量	用单引号 '''...'''	$PATH 不会被 Nextflow 替换，原样交给 Bash
需要同时用 Nextflow 变量和环境变量	用双引号 """..."""，环境变量前加 \ 转义	\$DB 交给 Bash，$MAX 由 Nextflow 替换

单引号字符串里，$PATH 不会被 Nextflow 处理，直接原样传给 Bash，由 Bash 在执行时解析。

process echo_path {
  '''
  echo "The path is: $PATH"
  '''
}

而更复杂的写法是

process blast {
  """
  blastp -db \$DB -query query.fa -outfmt 6 > blast_result
  cat blast_result | head -n $MAX | cut -f 2 > top_hits
  blastdbcmd -db \$DB -entry_batch top_hits > sequences
  """
}

• \$DB → 加了反斜杠，Nextflow 不替换，原样交给 Bash，由 Bash 在执行时读取环境变量 $DB
• $MAX → 没有转义，Nextflow 在执行前把它替换成实际的 Groovy 变量值

2. 脚本自选

Nextflow 的 Scripts à la carte（脚本自选） 机制：process 默认按 Bash 解释，但你完全可以用 Python、Perl、R 等其他语言，甚至在同一套流程里混用

一个管道可能由执行不同任务的进程组成。使用Nextflow，您可以选择最适合给定进程执行的任务的脚本语言。例如，对于某些进程，R可能比Perl更有用，而对于其他进程，您可能需要使用Python，因为它提供了更好的库或API访问等。

要点	说明
默认语言	Bash
切换语言	在 script 第一行加对应语言的 shebang（#!）
混用支持	不同 process 可以用不同语言，互不影响
适用场景	哪个语言更适合当前任务就用哪个，比如 R 做统计、Python 调 API

一个管道可能由执行不同任务的进程组成。使用Nextflow，您可以选择最适合给定进程执行的任务的脚本语言。例如，对于某些进程，R可能比Perl更有用，而对于其他进程，您可能需要使用Python，因为它提供了更好的库或API访问等。

process perl_task {
    """
    #!/usr/bin/perl

    print 'Hi there!' . '\n';
    """
}

process python_task {
    """
    #!/usr/bin/python

    x = 'Hello'
    y = 'world!'
    print "%s - %s" % (x,y)
    """
}

workflow {
    perl_task()
    python_task()
}

3. Conditional scripts

script部分就像一个返回字符串的函数。这意味着你可以编写任意代码来确定脚本内容，只要最终语句是一个字符串即可。

script 可以理解为一个生成 Shell 命令字符串的函数。你可以先执行一些 Groovy 代码，动态决定最终要运行什么命令，只要最后返回的是一个字符串即可

还可以使用if-else 命令

mode = 'tcoffee'

process align {
    input:
    path sequences

    script:
    if( mode == 'tcoffee' )
        """
        t_coffee -in $sequences > out_file
        """

    else if( mode == 'mafft' )
        """
        mafft --anysymbol --parttree --quiet $sequences > out_file
        """

    else if( mode == 'clustalo' )
        """
        clustalo -i $sequences -o out_file
        """

    else
        error "Invalid alignment mode: ${mode}"
}

把 process 中的 Shell 脚本单独放到一个文件里，让流程定义和实际执行命令分开，方便复用、测试和维护

以前把命令直接写在 Nextflow 文件里，现在把命令放到独立的 .sh 文件中，再让 Nextflow 调用它

流程脚本可以导出为模板文件，这样就可以在不同的流程中重复使用，并且可以独立于管道执行进行测试。

process hello {
    input:
    val STR

    output:
    stdout

    script:
    template 'hello.sh'
}

workflow {
    hello('Hello!').view()
}

• 默认情况下，Nextflow 会在定义流程的 Nextflow 脚本所在的模板目录中查找模板脚本。虽然可以使用绝对路径指定不同的位置，但不建议这样做，因为它会降低流程的可移植性。

hello.sh内容如下

#!/bin/bash
echo "process started at `date`"
echo $STR
echo "process completed"

注记

模板文件放在哪里、变量由谁解析、官方不太推荐使用模板. 最容易混淆的是：同一份 .sh 文件，既可以直接用 Bash 执行，也可以由 Nextflow 调用，但 $STR 的处理方式不同

4. Shell

当一条 Shell 命令里同时包含 Nextflow 变量和 Bash 变量时，可以用不同的符号区分它们，减少转义问题

shell: 从 Nextflow 25.04 起已经弃用（deprecated），新代码建议使用 script，script写法能够区分shell和groovy

process hello {
    input:
    val str

    shell:
    '''
    echo "User $USER says !{str}"
    '''
}

workflow {
    channel.of('Hello', 'Hola', 'Bonjour') | hello
}

在上面的示例中，$USER 被视为 Bash 变量，而 !{str} 被视为 Nextflow 变量,下面是简单的规则

• 使用 ! 的变量必须加花括号
• Shell 脚本必须使用单引号定义字符串
• Shell 支持 Template 机制

5. Native execution

exec部分会执行给定的代码，但不会启动作业。理解它的关键是：exec 可以直接执行 Groovy 代码，而不需要启动外部 Shell 进程。

process hello {
    input:
    val name

    exec:
    println "Hello Mr. $name"
}

workflow {
    channel.of('a', 'b', 'c') | hello
}

对比项	exec	script	shell
执行内容	Groovy 代码	Shell 脚本	Shell 脚本
是否启动外部 Shell	否	是	是
主要用途	简单计算、轻量级逻辑	调用外部程序、执行命令	区分 Nextflow 与 Bash 变量
Nextflow 变量引用	Groovy 语法，如 $name	通常使用 ${name}	使用 !{name}
Bash 变量引用	不直接支持 Bash 变量展开	$VAR，注意字符串插值与转义	$VAR
字符串定义	Groovy 字符串语法	通常使用 """..."""	通常使用 '''...'''
调用 Python / R / Linux 命令	需要显式调用外部程序	支持	支持
并行调度与任务缓存	支持	支持	支持
典型场景	轻量级数据处理	FastQC、STAR、fastp 等	同时使用 Bash 与 Nextflow 变量

6. Stub

给一个 process 准备一份“模拟执行脚本”，用来代替真正的分析命令，快速测试工作流能不能跑通

正在构建 RNA-seq 流程，真正的 Salmon index 可能需要处理很大的转录组文件，耗时较长。使用 stub，就可以不运行真正的索引程序，而是创建一些模拟输出文件，让下游进程继续运行

process salmon_index {
    input:
    path transcriptome

    output:
    path 'index'

    script:
    """
    salmon index --threads $task.cpus -t $transcriptome -i index
    """

    stub:
    """
    mkdir index
    touch index/seq.bin
    touch index/info.json
    touch index/refseq.bin
    """
}

• 真正运行 Salmon，对转录组进行索引，生成真实的索引文件
• sub 只创建一个 index 目录和几个空文件，不执行 Salmon

使用的时候用nextflow run main.nf -stub-run运行,就是模拟模式

3.2 Inputs

Input部分定义了进程的输入，类似于函数参数。一个进程最多只能有一个输入部分，该部分必须至少包含一个输入声明。

输入部分遵循以下语法：

input:
  <input qualifier> <input name>

输入声明由限定符和名称组成。输入限定符定义了要接收的数据类型。Nextflow 使用此信息来应用与每个限定符关联的语义规则，并根据目标执行平台（网格、云等）正确处理数据。

在工作流中调用流程时，必须为流程输入部分中的每个输入提供通道或数据流值，这类似于使用特定参数调用函数。以下各节中的示例演示了如何调用流程

重要

每个进程都要声明自己需要什么类型的数据，以及用什么变量名接收；调用进程时，必须为每个输入提供对应的数据

• Qualifier（限定符）：规定输入数据的类型，以及 Nextflow 应该如何处理它
• Name（名称）：给这个输入指定一个变量名，供进程内部使用

Input qualifier	作用说明
val	在进程脚本中通过变量名访问输入值
path	将输入值作为路径处理，并在执行环境中正确暂存文件
env	使用输入值设置进程脚本中的环境变量
stdin	将输入值传递给进程的标准输入（stdin）
tuple	将多个具有上述限定符的输入值组合成一个元组
each	针对输入集合中的每个元素分别执行进程

1. 输入变量(Val)

val限定符接受任何数据类型。可以在进程脚本中使用指定的输入名称访问它，如下例所示：

process echo {
  input:
  val x

  script:
  """
  echo "process job $x"
  """
}

workflow {
  def num = channel.of(1,2,3)
  echo(num)
}

在上面的示例中，该进程执行三次：对于num通道发出的每个值执行一次。结果输出类似于下面所示：

process job 3
process job 1
process job 2

注记

Nextflow 最核心的特性之一：异步并行处理（Asynchronous Parallel Execution）,进程处理项目的顺序，并不一定等于它们被接收（或输入）的顺序。在上面的例子中，数值 3 在其他数值之前就被处理完成了

2. Input files (path)

path 限定符用来把文件作为输入传给进程。Nextflow 会把这些文件 staging 到进程的工作目录里，然后你在 script: 块里就可以用你给这个输入起的名字来访问它

不是“把原始文件本身搬过去”，而是把文件“放进进程的工作目录里”，默认通常用软链接

process blast {
  input:
  path query_file

  script:
  """
  blastp -query ${query_file} -db nr
  """
}

workflow {
  def proteins = channel.fromPath( '/some/path/*.fa' )
  blast(proteins)
}

-channel.fromPath 是 workflow 里创建通道的方法，用来把磁盘上的文件路径变成一个 Nextflow 通道 - path query_file 是 process 里的输入限定符

在上面的例子中，所有以 .fa 结尾的文件都会被发送到 proteins 通道。这些文件会被进程接收，进程会对每个文件执行一次 BLAST 查询

channel.fromPath('/some/path/*.fa') 会匹配 /some/path/ 目录下所有 .fa 文件, 然后每个文件都会触发一次 blast process

在某些情况下，您的任务可能需要使用名称固定的文件，即它不必随实际提供的文件一起更改。在这种情况下，您可以在输入文件参数定义中用name属性指定一个固定的名称

input:
path query_file, name: 'query.fa'

input是输入文件路径，名字可以按照自己的要求来

path 限定符的另一种用法：它不仅能接收 channel.fromPath 产生的 file object，也能接收 字符串形式的绝对路径

1. file object

workflow {
    def proteins = channel.fromPath('/some/path/*.fa')
    blast(proteins)
}

• channel.fromPath 返回的是 Nextflow 的 file object
• 当它传给 process 的 path query_file 时，Nextflow 会把这个文件 staging 到任务工作目录里

2. 绝对路径

process cat {
  input:
  path x

  script:
  """
  cat $x
  """
}

workflow {
  cat('/some/data/file.txt')
}

path query_file 可以把它当作文件路径处理，并 staging 到任务目录里。

注记

当你给 path 输入加上 name: 参数时，文件会被 staging 到子目录里，此时 x.name 返回的是“带父目录的路径”，而不是单纯的文件名

input:
path x, name: 'my-dir/file.txt'

name: 不是“给变量起别名”，而是告诉 Nextflow：把这个文件 staging 到工作目录里的 my-dir/file.txt 这个位置

3. Multiple input files

path 输入不仅能接收单个文件，也能接收一组文件；这时输入变量会变成一个列表

process blast {
    input:
    path 'seq'

    script:
    """
    echo seq*
    """
}

workflow {
    def fasta = channel.fromPath( "/some/path/*.fa" ).buffer(size: 3)
    blast(fasta)
}

• buffer 是一个通道操作符，作用是把流里的元素按数量打包,buffer(size: 3) 表示：每收到 3 个元素，就打包成一个列表再发出去,比如a.fa, b.fa, c.fa, d.fa, e.fa, f.fa，变成[a.fa, b.fa, c.fa]，[d.fa, e.fa, f.fa]

• path 'seq' 会把这 3 个文件 staging 成类似seq_0,seq_1, seq_2

接收数量 (Arity)	名称模式 (Name pattern)	Staged 文件名结果
任意	*	保留源文件原本的名称
单个 (one)	file*.ext	file.ext
单个 (one)	file?.ext	file1.ext
单个 (one)	file??.ext	file01.ext
多个 (many)	file*.ext	file1.ext, file2.ext, file3.ext, …
多个 (many)	file?.ext	file1.ext, file2.ext, file3.ext, …
多个 (many)	file??.ext	file01.ext, file02.ext, file03.ext, …
多个 (many)	dir/*	保留源文件名，并放入 dir 子目录中
多个 (many)	dir??/*	保留源文件名，并放入递增索引的子目录中（如 dir01/, dir02/ 等）
多个 (many)	dir*/*	同上（放入递增索引的子目录中）

path 输入的一个校验功能：arity 选项。它用来强制规定这个输入到底应该收到几个文件

input:
path('one.txt', arity: '1')         // 必须恰好 1 个文件
path('pair_*.txt', arity: '2')      // 必须恰好 2 个文件
path('many_*.txt', arity: '1..*')   // 至少 1 个文件，上不封顶

任务启动时，Nextflow 会检查实际收到的文件数量是否匹配声明的 arity：

• 匹配：正常执行
• 不匹配：任务直接失败，报错提示文件数量不符

4.Dynamic input file names

文件名可以用同一个 process 的其他输入值来动态拼接

process grep {
  input:
  val x
  path "${x}.fa"

  script:
  """
  cat ${x}.fa | grep '>'
  """
}

• val x：接收一个字符串值，比如 “sample1”
• path "${x}.fa"：接收一个文件，但文件名由 x 决定

5. Input environment variables (env)

env 限定符：把输入值直接变成脚本里的环境变量

process echo_env {
    input:
    env 'HELLO'

    script:
    '''
    echo "$HELLO world!"
    '''
}

• env 'HELLO' 不是变量名，是环境变量的名字,通道发来什么值，HELLO 这个环境变量就等于什么值。脚本里用 $HELLO 就能取到。

• 脚本必须用单引号''', 因为 $HELLO 是留给 bash 在运行时展开的。如果用双引号 """，Groovy 会在提交任务前就尝试解析 $HELLO，那时它还不存在，会报错或展开为空。

6. Standard input (stdin)

stdin 限定符：把通道发来的值直接喂进脚本的标准输入（stdin），而不是通过文件或变量传递

process cat {
  input:
  stdin

  script:
  """
  cat -
  """
}

workflow {
  channel.of('hello', 'hola', 'bonjour', 'ciao')
    | map { v -> v + '\n' }
    | cat
}

• stdin 没有写变量名，因为它不绑定到某个变量，而是直接把值流式注入到任务脚本的 stdin
• map { v -> v + '\n' } 这一步很关键——如果不加换行，cat 输出时所有字符串会粘成一行helloworld!hola...。加了 \n 后每个值独立成行

map 是 channel 的转换操作符，意思是：channel 每发出一个元素，就用后面这个闭包处理一次，然后把处理后的结果继续往下游发

7. Input tuples (tuple)

tuple 限定符：把多个输入值打包成一个组合，一次接收

process cat {
    input:
    tuple val(id), path('input.txt')

    script:
    """
    echo "Processing $id"
    cat input.txt > copy
    """
}

workflow {
  channel.of( [1, 'alpha.txt'], [2, 'beta.txt'], [3, 'delta.txt'] ) | cat
}

• 这个 process 每次接收一个元组（一组值）
• 元组第一个元素是 val 类型，绑定到变量id
• 元组第二个元素是 path 类型，文件名是 input.txt

tuple 内部可以组合任意限定符：

• val — 普通值
• env — 环境变量
• path — 文件路径
• stdin — 标准输入

8. Input repeaters (each)

each 限定符：它的作用是让 process 对某个集合里的每个元素都执行一次，形成”交叉组合”的效果。

process align {
  input:
  path seq
  each mode

  script:
  """
  t_coffee -in $seq -mode $mode > result
  """
}

workflow {
  sequences = channel.fromPath('*.fa')
  methods = ['regular', 'espresso', 'psicoffee']

  align(sequences, methods)
}

• sequences 是一个通道，每发出一个文件，触发一次”主循环”
• methods 是一个普通列表，被 each 修饰后，每次主循环都会对它展开

假设有 2 个 fasta 文件（a.fa, b.fa），实际执行的任务数 = 2 × 3 = 6 次

注记

each 不只是能展开字符串列表，也能展开文件列表。这里 libraries 是一个文件数组，each path(lib) 会让 process 对每个库文件都跑一次

9. Multiple inputs

一个 process 可以接收多个输入，但其中只能有一个是队列通道（queue channel），其他必须是数据流值（dataflow values）

类型	创建方式	行为
队列通道（queue channel）	channel.of(...)、channel.fromPath(...)	可被消费，值取完就没了
数据流值（dataflow value）	channel.value(...)	单例值，可被重复读取，不会被消费

process echo {
  input:
  val x
  val y

  script:
  """
  echo $x and $y
  """
}

workflow {
  x = channel.value(1)
  y = channel.of('a', 'b', 'c')
  echo(x, y)
}

• process 等待 x 和 y 都有值
• x 是 dataflow value，值为 1，每次都能读到
• y 是 queue channel，依次发出 ‘a’、‘b’、‘c’
• y 每发出一个值，就和 x 组合触发一次任务

两个都是队列通道时，Nextflow 会把它们按到达顺序配对，但通道是异步的，谁先发谁后发不确定

3.3 Output

output部分定义进程的输出，类似于函数返回。一个进程最多只能有一个输出节，该输出节必须包含至少一个输出声明。

• 作用：定义 process 的产出，类似函数的 return
• 数量限制：一个 process 最多只能有一个 output: 块
• 非空要求：output 块里至少要声明一个输出

	input	output
作用	接收上游数据	向下游发送数据
数量	可以有多个	只能有一个 output 块
块内声明	至少一个	至少一个
限定符	val、path、env、stdin、tuple、each	val、path、tuple、emit 等

因为 output 的本质是定义这个 process 的返回通道。多个 output 块会导致返回值语义混乱。但一个 output 块里可以声明多个输出项

格式

output:
  <output qualifier> <output name> [, <option>: <option value>]

限定符	作用	示例
val	发出一个变量值	val result
path	发出一个文件	path 'output.txt'
env	发出环境变量中的值	env 'RESULT'
stdout	发出任务的标准输出	stdout
tuple	发出多个值的组合	tuple val(x), path(f)
eval	发出脚本/命令的执行结果	eval "grep -c '^>' ${fasta}"

1. Output variables (val)

val限定符允许您输出流程中定义的任何Nextflow变量。

1. 输出输入变量

process echo {
  input:
  each x

  output:
  val x

  script:
  """
  echo $x > file
  """
}

workflow {
  methods = ['prot', 'dna', 'rna']
  received = echo(methods)
  received.view { method -> "Received: $method" }
}

2. 三种 val 来源

process cat {
  input:
  path infile

  output:
  val x                    // ① 脚本变量
  val 'BB11'              // ② 字面量
  val "${infile.baseName}.out"  // ③ 表达式

  script:
  x = infile.name
  """
  cat $x > file
  """
}

workflow {
  ch_dummy = channel.fromPath('*').first()
  (ch_var, ch_str, ch_exp) = cat(ch_dummy)

  ch_var.view { var -> "ch_var: $var" }
  ch_str.view { str -> "ch_str: $str" }
  ch_exp.view { exp -> "ch_exp: $exp" }
}

• channel.fromPath('*') 扫描当前目录所有文件，生成一个通道
• .first() 只取第一个文件，发出一个 Path 对象

输出	来源	值（假设 infile = test.txt）
val x	脚本变量 x = infile.name	"test.txt"
val 'BB11'	字面量	"BB11"
val "${infile.baseName}.out"	Groovy 表达式	"test.out"

2. Output files (path)

path限定符允许您输出进程生成的一个或多个文件,输出文件本身（文件对象），不是纯字符串路径

process random_number {
  output:
  path 'result.txt'

  script:
  '''
  echo $RANDOM > result.txt
  '''
}

workflow {
  numbers = random_number()
  numbers.view { file -> "Received: ${file.text}" }
}

3. Multiple output files

path 通配符（glob）匹配多个输出文件的用法

当 output 的path 里包含 * 或 ? 通配符时，Nextflow 会按 glob 模式匹配所有符合条件的文件，把它们打包成一个 List 发出

process split_letters {
    output:
    path 'chunk_*'

    script:
    """
    printf 'Hola' | split -b 1 - chunk_
    """
}

workflow {
    split_letters
        | flatten
        | view { chunk -> "File: ${chunk.name} => ${chunk.text}" }
}

• 这里 split 命令把 Hola 按每 1 字节切分，生成 4 个文件。path ’chunk_*’ 匹配到它们，发出一个包含 4 个 Path 对象的 List：[chunk_aa, chunk_ab, chunk_ac, chunk_ad]

• 因为默认发出的是一个 List，如果直接 view，只会打印一次（整个列表）。加上 flatten 后，List 被”展开”，每个文件单独发出

4. arity

用来强制校验输出文件数量，不匹配就报错

output:
    path('one.txt', arity: '1')         // 必须恰好 1 个文件
    path('pair_*.txt', arity: '2')      // 必须恰好 2 个文件
    path('many_*.txt', arity: '1..*')   // 至少 1 个文件

限定符	作用	示例	说明
path（动态名称）	输出文件名由输入变量或脚本上下文动态决定	path "${species}.aln"	文件名依赖 input 中的 val species，每个任务产生不同名称的文件
env	输出执行环境中定义的变量值	env 'RESULT'	脚本中 RESULT=$(seq 5) 设置环境变量，通道发出该变量的值
stdout	输出进程的标准输出内容	stdout	脚本中 echo "Hello world!" 的输出直接作为通道元素发出
eval（v24.04+）	输出任意 Shell 命令的执行结果	eval('echo Hello world!')	在任务 Shell 上下文中执行单行命令，捕获其 stdout；命令失败则任务失败

5. Output tuples (tuple)

tuples元组限定符将单个输出中的多个值作为元组输出

output:
    tuple val(species), path('result')

• tuple 把 species（字符串）和 result（文件）打包成一个元组 (species, result)
• 通道每次发出的是一个元组，而不是两个独立的值
• 下游用 input: tuple val(sp), path(res) 接收时，两个值会被自动解构

6. emit 命名输出

emit 命名输出，它让多输出的 process 可以按名字访问，而不是靠位置

用 emit 命名后，通过 process名.out.名字 访问，代码可读性大幅提升。尤其当输出很多时，out[0]、out[1] 很难记住哪个是哪个

process variant_pipeline {
    output:
    path '*.vcf.gz', emit: vcf
    path '*.vcf.gz.tbi', emit: tbi
    path '*.log', emit: log

    script:
    """
    gatk HaplotypeCaller -I input.bam -O sample.vcf.gz
    tabix sample.vcf.gz
    """
}

workflow {
    result = variant_pipeline()

    // 清晰明了
    result.out.vcf.view()
    result.out.tbi.view()
    result.out.log.view()
}

7. optional

optional: true 可选输出——允许 process 在某些情况下不产生某个输出文件而不报错

output:
    path("output.txt"), optional: true

• 不加 optional：如果任务没产生 output.txt，任务直接失败
• 加了 optional: true：如果 output.txt 不存在，任务不报错，但该输出通道不会发出任何值

3.4 Directives

Process 指令（Directives） 的总览。指令写在 input 之前，用来控制任务的执行方式、资源需求和软件依赖

核心特性：动态求值

指令默认在 process 定义时求值一次，但如果值是闭包（用 {} 包裹），则每个任务单独求值，可以使用 task 对象和输入变量

process align {
    input:
    tuple val(sample_id), path(reads)

    // 根据样本名动态设置内存
    memory { sample_id == 'large_sample' ? '16 GB' : '4 GB' }

    // 根据任务尝试次数动态调整策略
    errorStrategy { task.attempt < 3 ? 'retry' : 'ignore' }

    script:
    """
    bwa mem ref.fa $reads > aligned.bam
    """
}

类别	指令	作用
通用	errorStrategy	任务失败时的处理策略（retry、ignore、finish、terminate）
	executor	指定执行器（local、slurm、pbs 等）
	tag	给任务打标签，日志中显示更易读的名称
资源	cpus	申请 CPU 核心数
	memory	申请内存大小
	time	申请 walltime 时长
软件依赖	conda	指定 conda 环境/包
	container	指定容器镜像（Docker/Singularity）

1. Dynamic directives

动态指令（Dynamic directives）——让指令的值在任务执行时根据输入动态决定，而不是在 process 定义时就固定

核心机制：闭包

process hello {
    executor 'sge'
    queue { entries > 100 ? 'long' : 'short' }

    input:
    tuple val(entries), path('data.txt')

    script:
    """
    your_command --here
    """
}

代码	含义
process hello {	定义一个名为 hello 的 process
executor 'sge'	指定使用 SGE（Sun Grid Engine）集群执行器来提交任务
queue { entries > 100 ? 'long' : 'short' }	动态指令：用闭包 {} 包裹，每个任务执行时根据输入变量 entries 的值决定提交到哪个队列——entries > 100 走 long 队列，否则走 short 队列
input:	输入声明块开始
tuple val(entries), path('data.txt')	接收一个元组：entries 是一个数值（val），data.txt 是一个文件（path）
script:	脚本块开始
your_command --here	实际要执行的命令（占位符）

• executor 是静态的：所有任务都用 SGE 执行器，因为它没有用闭包包裹
• queue 是动态的：用闭包 {} 包裹，每个任务根据自身输入 entries 的值独立决定队列

这个 process 用 SGE 执行器提交任务，并根据输入变量 entries 的大小动态选择 long 或 short 队列——数据量大的任务走长队列，小的走短队列。

2. Dynamic task resources

用动态指令让每个任务根据自身需求申请资源，避免”一刀切”导致的资源浪费或任务失败

同一个 process 的不同任务，资源需求可能差异很大：

• 申请太少 → 部分任务 OOM 失败
• 申请太多 → 所有任务排队时间长，浪费资源

process hello {
    memory { 2.GB * task.attempt }
    time { 1.hour * task.attempt }

    errorStrategy { task.exitStatus in 137..140 ? 'retry' : 'terminate' }
    maxRetries 3

    script:
    """
    your_command --here
    """
}

代码	含义
memory { 2.GB * task.attempt }	动态内存：用闭包 {} 包裹，根据 task.attempt（当前尝试次数）动态计算。第 1 次申请 2 GB，第 2 次 4 GB，以此类推
time { 1.hour * task.attempt }	动态时长：同理，第 1 次 1 小时，第 2 次 2 小时……
errorStrategy { ... }	动态错误策略：如果退出码在 137–140 之间（通常是 OOM 被系统 kill），执行 retry（重试）；否则执行 terminate（立即终止整个流程）
maxRetries 3	最多重试 3 次（加上首次执行，最多跑 4 轮）

4. Workflows

process 是“计算节点”，workflow 是“数据流编排函数”

在 Nextflow 中，workflow 是一种专门用来组合 process 和数据流逻辑的函数。

• 入口工作流是流水线的入口点。它可以通过params 块接收参数作为输入，并通过 output 块发布输出

• 命名工作流是可以被其他工作流调用的工作流。它可以定义自己的输入和输出，分别称为 takes 和 emits

• 入口工作流和命名工作流都可以包含数据流逻辑，例如调用 process、调用其他 workflow，以及使用 channel operator

在 DSL2 里，workflow 是用来组合 process 和数据流的函数；没有名字的 workflow 通常是入口，负责 params 输入和 output 发布；有名字的 workflow 是子模块，通过 takes 接收输入、通过 emits 提供输出。

4.1 Entry workflow

一个最小入口工作流：脚本运行时，Nextflow 会自动执行这个无名的 workflow { … } 块

workflow {
    channel.of('Bonjour', 'Ciao', 'Hello', 'Hola')
        .map { v -> "$v world!" }
        .view()
}

注记

entry workflow 不是”简单的 workflow”，它的核心作用是：给整个脚本一个自动执行的入口，并把参数入口和输出出口统一收拢在这里

因为 Nextflow 脚本里可以写很多 process、很多 workflow、很多 channel，但引擎需要知道：

用户执行 nextflow run xxx.nf 时，到底从哪里开始跑？

作用	说明
入口	脚本运行时自动执行，不需要被显式调用
接参数	通过 params 接收外部输入，比如输入文件、输出目录
编排数据流	把 channel、process、子 workflow 连起来

4.2 Parameters

Parameters 理解成 Nextflow pipeline 的“外部配置入口”, 如果没有 params，你每次换输入文件、换输出目录、换开关，都得改脚本。

params.input = '/some/data/file'
params.save_intermeds = false

workflow {
    if( params.input )
        analyze(params.input, params.save_intermeds)
    else
        analyze(fake_input(), params.save_intermeds)
}

4.3 Output

Nextflow 25.10 正式稳定的 output 块 / workflow outputs，它的核心作用是：在入口 workflow 里声明“哪些 channel 是 pipeline 最终输出”，并决定这些文件从 work 目录发布到哪里

层级	作用	示例
process output	声明 process 产生哪些文件	output: path 'sample.txt'
workflow publish:	把某个 channel 标记为要发布的输出	publish: samples = ch_samples
output { ... }	声明这个 workflow 输出的发布路径	samples { path '.' }

process fetch {
    output:
    path 'sample.txt'
}

workflow {
    main:
    ch_samples = fetch(params.input)

    publish:
    samples = ch_samples
}

output {
    samples {
        path '.'
    }
}

• fetch 产生 sample.txt,这个 process 运行后，会在 work 目录里产生 sample.txt 这个文件，并把它放进输出 channel
• 给 ch_samples 这个 channel 起一个“输出名字”，叫 samples,它不是“把 channel 复制到磁盘”, 把 ch_samples 这个 channel 登记成 workflow 的一个输出，名字叫 samples,只有登记过的输出，外层的 output {} 才能配置它

部分	含义
ch_samples	一个真实的 channel，里面装着 process fetch 的输出
samples	这个 workflow 输出的名称，相当于一个标签
samples = ch_samples	把 ch_samples 这个 channel 注册为名为 samples 的 workflow 输出

• output { samples { path '.' } } 表示这些文件发布到输出目录的根路径

旧写法是在每个 process 里写 publishDir

fetch 产生 sample.txt
        ↓
放进 ch_samples
        ↓
publish: samples = ch_samples
        ↓
output { samples { path '.' } }
        ↓
从 work 目录发布到 results/

1. Publishing files

每个工作流输出都可以定义如何将文件从工作目录发布到指定的输出目录

2. Output directory

可以使用-output-dir命令行选项或outputDir配置选项设置运行的顶级输出目录：

nextflow run main.nf -output-dir 'my-results'

// nextflow.config
outputDir = 'my-results'

work/          → 任务执行目录，存放中间结果
results/       → 最终输出目录，由 output {} 控制

• work 目录不是最终输出目录, process 在 work 目录里执行，是因为 Nextflow 把每个任务隔离成独立执行空间；而最终文件要不要放到 results/，则由 workflow 的 output {} 或 publishDir 决定

3. Dynamic publish path

path指令也可以是一个闭包，它为每个通道值定义一个自定义的发布路径：

workflow {
    main:
    ch_samples = channel.of(
        [id: 'SAMP1', fastq_1: file('1.fastq'), fastq_2: file('2.fastq')]
    )

    publish:
    samples = ch_samples
}

output {
    samples {
        path { sample -> "fastq/${sample.id}/" }
    }
}

4. index

Index 文件：给输出建一份“清单”

output {
    samples {
        path 'fastq'
        index {
            path 'samples.csv'
        }
    }
}

它的作用是：把 channel 里的结构化信息保存成一个 CSV / JSON / YAML 文件。

5. Label

label 不是改文件路径，而是给这个 workflow 输出附加元数据。可以通过 data lineage 查询，比如“找出所有标记为 qc 的输出”

output {} 里除了 path、index、label，还可以写一组“发布行为配置”

指令	作用
enabled	是否启用这个 output 的发布规则
mode	发布方式，如 copy、link、move
overwrite	目标已存在时是否覆盖
includeInputs	是否发布来自输入而非 work 目录的文件
ignoreErrors	发布出错时是否忽略
contentType	设置发布文件的 MIME 类型
storageClass	云存储场景下的存储类别
tags	附加标签，常用于云存储或追踪

4.4 Named workflows

process 只能封装”单个计算步骤”, 想把 多个 process / channel 逻辑 打包成一个模块

workflow my_workflow {
    ch_hello = hello()
    bye( ch_hello.collect() )
}

workflow {
    my_workflow()
}

写法	角色
workflow { ... }	入口 workflow，没有名字，脚本从这里开始执行
workflow my_workflow { ... }	具名 workflow，相当于一个可复用的子工作流

• hello() 是一个 process 或子 workflow
• 它的输出变成 ch_hello
• ch_hello.collect() 把流里的元素收集起来
• 再传给 bye()

1. take

take: 就是 具名 workflow 的输入参数声明，用来告诉这个子 workflow：我运行时需要从外面拿到哪些 channel

workflow my_workflow {
    take:
    data1
    data2

    main:
    ch_hello = hello(data1, data2)
    bye(hello.out)

}

data1、data2 就是 my_workflow 的两个输入, 把 data1 和 data2 传给 hello，并将 hello 的返回输出赋给 ch_hello

调用的时候怎么传进去

workflow {
    my_workflow( channel.of('/some/data') )
}

调用 my_workflow(…) 时，括号里的参数会按顺序绑定到 take: 里声明的名字：

• 第 1 个参数 → data1
• 第 2 个参数 → data2

2. emit

子 workflow 把内部 channel 作为返回值暴露给外层,给 workflow 的输出定义名称，让外部 workflow 可以访问这些输出。

workflow my_workflow {
    main:
    ch_hello = hello(data)
    ch_bye = bye(ch_hello)

    emit:
    hello = ch_hello
    bye = ch_bye
}

外层调用

result = my_workflow()
result.hello.view()
result.bye.view()

4.5 Dataflow

工作流由数据流逻辑组成，其中进程通过数据流通道和数据流值相互连接。

1. Channels and values

• Channel：一条可以不断传递多个数据的流水线
• Dataflow Value：一个只会产生一次结果的数据容器

两者都可以把数据传递给 Nextflow 的 operator 或 process，而不是像普通变量一样直接读取其内部数据。

channel.of(1, 2, 3).view { v ->
    "channel emits ${v}"
}

• v 不是整个 Channel，而是 Channel 每次发出的单个值,也是 Nextflow 能够并行处理多个样本的基础。例如，一个 Channel 传入 100 个 FASTQ 文件，就可以将这些数据逐个交给后续的 process 处理

数据流值

channel.value(1).view { v ->
    "dataflow value is ${v}"
}

这里的 channel.value(1) 创建的是一个 Dataflow Value，而不是普通的 Channel。 它和 Channel 的主要区别是：

特征	Channel	Dataflow Value
数据数量	可以发出多个值	只提供一个值
是否异步	是	是
数据何时可用	随数据流逐步到达	值确定后可用
典型用途	多个样本、多个文件	一个参数、一个共享配置值
创建方式	channel.of(...)	channel.value(...)

2. Factories

Nextflow 中的 Factories（工厂方法）：通过 channel 命名空间里的不同方法，创建 Channel 并向其中放入数据.

Factory 就是创建 Channel 的方法，不同的 factory 可以用不同的方式生成数据流。

channel.fromPath('input/*.txt').view()

会创建一个包含匹配文件路径的Channel,*.txt 是通配符，表示匹配该目录下以 .txt 结尾的文件

常见的factory方法

Factory	用途	示例
channel.of()	根据给定的值创建 Channel	channel.of(1, 2, 3)
channel.value()	创建一个 Dataflow Value	channel.value(10)
channel.fromPath()	根据文件路径或通配符查找文件	channel.fromPath('data/*.fastq.gz')
channel.fromList()	从列表创建 Channel	channel.fromList([1, 2, 3])
channel.fromFilePairs()	按命名模式匹配成对的文件，常用于双端测序数据	channel.fromFilePairs('data/*_{1,2}.fastq.gz')

注记

Channel Factory 通常在 workflow 中使用，而不是直接在 process 中使用。因为它的主要作用是创建 Channel，为后续的 process 提供输入数据

3. Operators（操作符）

如果说 Factory 是用来创建 Channel 的，那么 Operator 就是用来处理、转换、筛选和组合 Channel 中的数据的

• Factory：产生数据流
• Operator：加工和组织数据流
• Workflow：组织整个工作流程
• Process：执行实际的计算任务

Operator 接收数据流，处理数据，然后产生新的数据流。

Operator	作用	简单理解
collect	将 Channel 发出的值收集成一个集合	多个值 → 一个集合
combine	组合两个 Channel 的值，生成组合结果	两组数据进行组合
filter	只发出满足条件的值	筛选数据
flatMap	将每个输入值转换成零个、一个或多个输出值	一个值 → 多个值
groupTuple	根据 Tuple 中的键对数据分组	按样本 ID 等键分组
join	根据匹配的键连接两个 Channel	按 ID 匹配数据
map	转换每个输入值	一个值 → 一个转换后的值
mix	将多个 Channel 的值合并到一个 Channel	多路数据汇入一路
view	打印 Channel 发出的值	调试和查看数据

4. Calling processes and workflows

Nextflow 中 Process 和 Workflow 的调用规则、返回值、输出访问方式，以及如何重复调用同一个模块,之前学的 take、main、emit、Channel 和 Operator，可以把这段理解为：Nextflow 如何像调用函数一样，把一个任务或工作流的输入传进去，再把输出接出来

workflow负责调用Process

workflow hello_bye {
    take:
    data

    main:
    hello()
    bye(data)
}

workflow {
    data = channel.fromPath('/some/path/*.txt')
    hello_bye(data)
}

在 Nextflow 的标准调用结构中，不能把 Process 当作普通函数，随意在脚本的任意位置执行。需要通过 Workflow 来组织和调用它

同一个 Process 或 Workflow，在同一个 Workflow 中只能调用一次

如何重复使用同一个process或workflow

include { hello_bye as hello_bye1 } from './modules/hello_bye'
include { hello_bye as hello_bye2 } from './modules/hello_bye'

workflow {
    data1 = channel.fromPath('data1/*.txt')
    data2 = channel.fromPath('data2/*.txt')

    hello_bye1(data1)
    hello_bye2(data2)
}

• 调用 Process，返回的是它声明的输出
• 调用 Workflow，返回的是它通过 emit: 暴露的输出

定义位置	输出命名方式	调用后访问
Process	output: path ..., emit: txt	hello_out.txt
Workflow	emit: hello = ...	flow_out.hello

5. Special operators (| and &)

Nextflow 中两种特殊运算符：|（管道）和 &（分叉）。它们可以简化 Workflow 里的数据流连接，让代码更像一条流水线。

1. |：管道运算符（Pipe）

把前一步产生的数据，直接传给下一步

channel.of('Hello')
    | map { v -> v.reverse() }
    | view

这里的 |不是 Linux Shell 中的管道，而是 Nextflow 在 Workflow 中用于连接 Channel、Operator、Process 和 Workflow 的特殊语法。

2. &：分叉运算符（Fork）

将同一份输入数据，同时交给多个 Process 并行处理

channel.of('Hello')
    | (greet & to_upper)

• greet：在字符串后面添加 world。
• o_upper：将字符串转换为大写。

6. recursion

核心思想是：一个 process 或 workflow 执行完后，把本次输出重新作为下一次执行的输入，反复运行，直到满足停止条件。

5.Static typing

静态类型允许您在数据流经管道时对其结构进行建模和验证。您可以将标准的Nextflow类型添加到每个级别的输入和输出，从管道参数到流程和工作流输出。Nextflow使用这些注释在编译时捕获与类型相关的错误。

静态类型是可选的。所有现有的代码都可以继续工作

1. Static typing features

原文	含义	怎么理解
Typed processes	process 的输入输出可以写类型注解	比如 Path、String、Map、Channel<Map>，而不再只靠 val、path 这些限定符
Typed workflows	workflow 的 take、emit 可以写类型	能明确 workflow 接收什么、发出什么，比如 Channel<String>
Typed parameters	params 块里的参数可以写类型	比如 input: Path、save_intermeds: Boolean = true
Typed outputs	pipeline 输出可以写类型	用 output { ... } 声明最终输出 channel 的结构
Operators	推荐配合静态类型使用的 channel 算子	比如 map、filter、groupTuple 等，让 channel 结构更清晰

2. Enabling static typing

启用静态类型,两种方式：

• 严格解析器：NXF_SYNTAX_PARSER=v2, 输入export NXF_SYNTAX_PARSER=v2
• 脚本里启用类型：nextflow.enable.types = true,用了 typed process 或 typed workflow 的脚本里nextflow.enable.types = true

3. Typed processes

类型化流程对输入和输出使用支持静态类型的新语法。

nextflow.enable.types = true

process hello {
    input:
    message: String

    output:
    file('hello.txt')

    script:
    """
    echo '${message}' > hello.txt
    """
}

1. Inputs

input：节声明流程输入。在类型化进程中，每个输入声明由名称和类型组成：

nextflow.enable.types = true

process fastqc {
    input:
    meta: Map
    fastq: Path
    extra_args: String

    script:
    """
    echo 'meta: ${meta}'
    echo 'fastq: ${fastq}'
    echo 'extra_args: ${extra_args}'
    """
}

写法	含义
map { v -> v + 1 }	channel 算子，对每个元素做转换
meta: Map	输入类型注解，表示这个输入是 Map 类型

2. File inputs

nextflow.enable.types = true

process cat {
    input:
    reads: List<Path>

    script:
    """
    cat ${reads.join(' ')}
    """
}

代码	含义
nextflow.enable.types = true	启用 typed process / typed workflow
reads: List<Path>	输入 reads 必须是一个 Path 列表
reads.join(' ')	把列表里的路径用空格拼成一个字符串
cat ${reads.join(' ')}	最终执行类似 cat file1 file2 file3

3. Nullable inputs

默认情况下，如果任何输入接收到空值，任务将失败。要允许空值，添加？到类型注释：

nextflow.enable.types = true

process cat_opt {
    input:
    input: Path?

    stage:
    stageAs input, 'input.txt'

    output:
    stdout()

    script:
    '''
    [[ -f input.txt ]] && cat input.txt || echo 'empty input'
    '''
}

静态类型写法	对应旧语法 / 等价含义	说明
name: Path	path name	普通文件/目录输入，会自动 stage 到 task 目录
name: Path?	可选 path name	允许输入为 null；默认情况下输入为 null 会导致任务失败
name: String	val name	普通字符串/数值等非文件输入
name: List<Path>	多个 path 或列表输入	接收一组文件路径
name: Map	val name 接收 Map	接收键值对集合
sample: Sample	tuple val(...), path(...) 的封装	使用 record 类型接收结构化输入；Path 字段会自动 stage
record(id: String, fastq: Path)	tuple val(id), path(fastq)	解构 record 输入，只取需要的字段
tuple(id: String, fastq: Path)	tuple val(id), path(fastq)	解构 tuple 输入，每个字段按类型 stage
env 'HELLO', hello	env hello	把输入值设置成脚本中的环境变量
stdin message	stdin message	把输入值送入脚本的 stdin
stageAs fasta, 'query.fa'	path 'query.fa'	自定义文件 stage 名称/路径
stageAs fasta, "${id}.fa"	动态命名输入文件	用其他输入变量决定 stage 后的文件名

4. Typed workflows

nextflow.enable.types = true

workflow hello_bye {
    take:
    samples: Channel<Path>

    main:
    ch_hello = hello(samples)
    val_bye = bye(ch_hello.collect())

    emit:
    result: Value<Path> = val_bye
}

• 类型化工作流 (Typed Workflows): 允许在命名工作流的 take:（输入）和 emit:（输出）部分声明数据类型。Nextflow 会据此验证进出工作流的值。

• 流水线组合 (Pipeline Composition): 功能：允许将整个流水线（包括 params、入口工作流和 output 块）作为一个命名工作流包含并调用。

5. Typed parameters

方面	说明
params 块	用 params { name: Type = default } 声明参数、类型和默认值
类型支持	支持 Path, String, Integer, Boolean, Map, List, Record，以及 Channel<E>、Value<V>
启用方式	params 块本身不需要 nextflow.enable.types = true；类型化 workflow/process 才需要
必填/可选	无默认值通常为必填；有默认值则为可选
Boolean 特殊规则	26.04 起，没有默认值的 Boolean 参数默认取 false，不再视为必填
Path 参数	声明为 Path 的输入文件在启动时必须已经存在；流程内部生成的路径应使用 String
参数覆盖	命令行、params 文件、config 都可覆盖；多源冲突时按 Nextflow 参数解析顺序决定最终值
命令行转换	命令行传入的值会自动转换成声明类型，例如 Integer 参数不会保持为字符串
引用限制	类型化参数应只在入口 workflow 或 output 块中引用；其他地方应通过显式输入传递
校验能力	语言服务器可在编码阶段检查参数类型；运行时会校验 CLI 参数和参数引用

6. Notifications

本页说明如何处理工作流事件和发送通知。

1. Workflow Handlers（工作流处理器）

Nextflow 是异步执行的，脚本结束不等于工作流执行结束。因此，一些只有在执行完成后才能获取的信息（如执行状态、完成时间等），需要通过异步处理器来访问

• 当工作流执行完成时，Nextflow 会调用 onComplete 事件处理器
• 可以访问工作流的终止状态和其他有用信息

workflow {
    main:
    // ...

    onComplete:
    println "Pipeline completed at: $workflow.complete"
    println "Execution status: ${ workflow.success ? 'OK' : 'failed' }"
}

• workflow.complete：工作流完成的时间
• workflow.success：布尔值，表示执行是否成功

2. Error handler

当运行时错误或进程错误导致流水线停止执行时，Nextflow 会调用 onError 事件处理器

workflow.onError {
    println "Error: Pipeline execution stopped with the following message: ${workflow.errorMessage}"
}

• onError：在错误发生时立即被调用
• onComplete：在流水线即将终止之前被调用

3. Mail

使用sendMail函数从工作流脚本发送邮件消息。

sendMail {
    to 'you@gmail.com'
    from 'me@gmail.com'
    attach '/some/path/attachment/file.txt'
    attach '/other/path/image.png'
    subject 'Catch up'

    // string expression is treated as the mail body
    '''
    Hi there,
    Look! Multi-lines
    mail content!
    '''
}

• 默认行为如果未配置 SMTP，Nextflow 会尝试调用系统的外部命令如 sendmail 或 mail

Mail configuration

如果没有提供邮件服务器配置，Nextflow将尝试使用底层系统最终提供的外部邮件命令（例如sendmail或mail）发送邮件。

SMTP 配置：在 nextflow.config中显式配置 SMTP服务器

mail {
    smtp.host = 'smtp.yourcompany.com'
    smtp.port = 465          // 常见端口：25, 587, 465
    smtp.user = 'your_username'
    smtp.password = 'your_password' // 注意安全性
    smtp.ssl.enable = true   // 启用 SSL/TLS
}

4. 自定义邮件通知

• 灵活性高：你可以完全控制邮件的内容、格式和附件
• 发送包含详细执行摘要（如完成时间、持续时间、工作目录、退出状态等）的 HTML 或纯文本报告

workflow.onComplete {
    def summary = """...""" // 构建自定义消息体
    sendMail(to: 'user@example.com', subject: '...', body: summary)
}

注记

Nextflow 内置通知机制 (-N 参数)

nextflow run <pipeline_name> -N recipient@example.com

7. Secrets

Nextflow 的 Secrets（密钥/敏感信息管理） 功能，旨在解决生物信息学流程中账号密码、API Key 等敏感数据的存储和传递安全问题

• 解耦存储：密钥不与流水线代码或配置文件硬编码在一起，而是存储在本地安全的地方 only accessible to the owner
• 注入方式：运行时由 Nextflow 注入，作为 环境变量 (Environment Variables) 传递给 Task
• 安全性保证：不会泄露到临时执行文件或日志中

# create a new secret
nextflow secrets set FOO "Hello world"

# list all secrets
nextflow secrets list

# get the value of secret FOO
nextflow secrets get FOO

# delete the secret FOO
nextflow secrets delete FOO

1. 准备阶段：存储密钥

你需要将密钥保存到 Nextflow 的本地安全存储中。这只需要在终端执行一次命令

# 设置一个名为 MY_AWS_KEY 的密钥，值为 "your_actual_key_value"
nextflow secrets set MY_AWS_KEY "your_actual_key_value"

# 验证是否添加成功
nextflow secrets list

2. 在 Process 中使用（最常用方式）

通过 secret 指令，Nextflow 会将密钥作为环境变量注入到具体的 Task 中

process align_genome {
    // 1. 声明需要使用的密钥名称
    secret 'MY_AWS_KEY'
    secret 'MY_DB_PASSWORD'

    input:
    path genome_file

    output:
    path aligned_file

    script:
    """
    # 在脚本中直接使用环境变量引用密钥
    # 注意：在 Groovy 字符串中，Shell 变量需要用 \$ 转义，防止被 Groovy 解析
    aws s3 cp s3://my-bucket/data/ ./data/ --key \$MY_AWS_KEY
    
    mysql -u admin -p\$MY_DB_PASSWORD my_database < schema.sql
    """
}

3. 在配置文件 (nextflow.config) 中使用.

8. Sharing pipelines

Nextflow 确实原生支持并与主流 Git 平台（GitHub、GitLab、Bitbucket）深度集成，这构成了其可重现性（Reproducibility）和协作能力的核心基础

• Nextflow 脚本（.nf）、配置文件（.config）和资源文件都可以通过 Git 进行版本管理
• 可以精确地记录某个分析结果是基于哪个 commit hash 或分支运行的
• Nextflow 的 nextflow run 命令可以接受一个 Git URL 作为参数，自动克隆仓库并运行指定版本的流程

特性	GitHub	GitLab	Bitbucket
主要用户群	开源社区、全球开发者、科技公司	DevOps 团队、企业级用户	Atlassian 用户、中小团队、部分学术机构
优势	最大的开源社区，发现新工具最容易	强大的 CI/CD 内置功能，自托管选项丰富	与 Jira/Trello 无缝集成，私有仓库政策宽松（早期）
Nextflow 中的地位	目前最主流 (nf-core 项目主要在此)	越来越多地被采用	传统重要平台，许多旧流程和内部流程仍在此

注记

Nextflow并不打算完全取代Git工具,仍然需要git来创建新的存储库或提交更改

8.1 Git configuration

你可以通过编辑位于 ~/.nextflow/scm 的文件，来配置 Nextflow 访问不同 Git 平台（如 GitHub、GitLab、Bitbucket）时所需的身份验证凭证

告诉 Nextflow 你的账号密码或 Token，让它能自动下载代码

当你运行一个托管在私有仓库（Private Repository）或者需要认证的公共仓库中的 Nextflow 流程时，Nextflow 需要知道“你是谁”，才能从 Git 服务器拉取代码

可以创建一个文本文件 ~/.nextflow/scm:

[github]
    user = your_github_username
    token = ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx  # 你的 GitHub Personal Access Token

[gitlab]
    user = your_gitlab_username
    token = glpat-xxxxxxxxxxxxxxxxxxxx              # 你的 GitLab Private Token

8.2 Using a local repository

Nextflow 如何直接从本地或共享文件系统中的 Git 仓库运行流程，而无需连接到 GitHub、GitLab 等远程服务器

你可以把代码放在自己的电脑硬盘或公司内网服务器上，让 Nextflow 直接读取并运行。

注记

Bare repository（裸仓库）

• 这是 Git 的一种特殊仓库格式
• 普通仓库: 包含代码文件（.nf, .config）和一个隐藏的 .git 文件夹
• 裸仓库 (*.git): 只包含 Git 的历史记录数据库，不包含实际的工作文件

第一步：创建裸仓库

# 进入你的项目目录
cd my-nextflow-project

# 将其克隆为一个裸仓库到指定位置
git clone --bare . /shared/projects/hello.git

第二步：在 Nextflow 中运行它

nextflow run file:/shared/projects/hello.git

场景	说明
内网/离线环境	医院、政府机构或大型企业的服务器无法访问外网 GitHub，只能内部部署代码。
快速测试与开发	开发者在本地修改代码后，想立即在本机验证，无需推送远程再拉取。
团队内部协作	小团队通过 NFS 或 SMB 共享一个 Git 仓库，大家都能从同一个地方拉取最新代码。
私有数据保护	代码中包含敏感信息，绝对不能上传到公共 Git 平台。

8.3 Publishing your pipeline

如何将你的 Nextflow 流程（Pipeline）发布到 GitHub（或其他平台），以便让其他人能够轻松使用和运行你的代码

1. 发布的前提条件

• 创建仓库:你需要在 GitHub 上创建一个仓库（Repository）

• 包含文件: 这个仓库里必须包含你所有的脚本和数据文件

• 主脚本名称:

  • 默认情况下，Nextflow 要求主入口脚本命名为 main.nf
  • 定义名称: 如果你不想叫 main.nf，可以在项目的 nextflow.config 文件中通过 manifest.mainScript 属性指定其他名

// nextflow.config 示例
manifest {
    mainScript = 'my_custom_script.nf'
}

2. 他人如何运行你的流程

一旦你将项目上传到 GitHub，其他人可以通过以下两种方式之一来运行你的流程

# 使用简短的用户名/仓库名(推荐)
nextflow run acme/hello

# 使用完整的 GitHub URL
# 格式: nextflow run https://github.com/<用户名>/<仓库名>.git
nextflow run http://github.com/acme/hello

• 允许团队成员或全球开发者直接复用你写的分析流程，无需手动下载 ZIP 包
• 利用 GitHub 的版本历史功能，方便追踪流程和代码的变化

8.4 管理依赖

Nextflow 如何管理和解决“依赖项（Dependencies）”问题

• 代码：其他的辅助脚本、库
• 数据：参考基因组、索引文件等
• 环境：特定的软件版本（如 Python 3.9, BWA v0.7）、环境变量、计算资源（CPU/内存）

1. Code

Nextflow 提供了三种方式来包含外部工具或自定义脚本：

1. 标准依赖（Docker / Conda）—— 最推荐

不依赖用户本地安装了什么软件，而是直接拉取标准的容器镜像（Docker）或环境包（Conda/Spack）

docker.enabled = true
conda.enabled = true

在 process 中指定 container 或 conda 指令

2. bin 目录：自定义脚本

如果你有一些自己写的、非标准的脚本（比如 .py, .sh），可以放在仓库的 bin/ 文件夹下。

• Nextflow 会自动把它们加入 PATH，你可以像调用普通命令一样调用它们（无需写绝对路径）
• 如果修改了 bin 里的脚本，Nextflow 会检测到变化并重新执行相关任务（缓存失效机制）

使用 #!/usr/bin/env python 作为 shebang，而不是硬编码路径（如 #!/usr/bin/python），以提高跨平台兼容性

3. lib 目录：Groovy 类库

如果你需要在 Nextflow 的主脚本（.nf）中复用复杂的逻辑（比如序列处理算法），可以将 Groovy 类放在 lib/ 目录下

• Nextflow 会自动将 lib/ 添加到 classpath
• 在 .nf 脚本中直接 new ClassName() 即可实例化。

2. Data

Nextflow 根据数据的性质（大小、变化频率），推荐了两种不同的处理方式：

• 外部输入数据（通过 Parameters 参数化）
  • 适用场景：大型数据集（如测序原始数据 FASTQ/BAM）、经常变化的文件

代码与数据分离”。流程脚本本身不包含具体数据，而是通过“参数”接收数据路径。这样同一个流程可以处理成千上万不同的样本

• 内部小型数据（内置于仓库）
  • 适用场景极小且固定不变的数据：如参考基因组的小片段、特定的索引文件、配置文件等 > 为了方便分发，将少量必要数据直接打包进 Git 仓库

$projectDir 是一个内置隐式变量，指向当前项目根目录

3. Environment

“环境”不是指 Docker 容器或 Conda 环境，而是指 操作系统层面的配置 和 计算资源调度

1. 环境变量 (Environment Variables)

如果某个工具依赖特定的环境变量（例如 AWS 的密钥、特定的库路径 LD_LIBRARY_PATH），可以在 nextflow.config 中统一声明。

2. 执行器配置 (Executors / Profiles)

当你把流程提交到高性能计算集群（HPC，如 SLURM, PBS）或云平台（AWS Batch, Azure Batch）时，你需要告诉 Nextflow 如何向调度器发送任务