Processes（进程）

我们现在知道如何创建和使用 Channels 来发送数据。现在我们将看到如何使用进程在工作流中运行任务。
进程是 Nextflow 执行的是你在命令行上运行的命令或自定义脚本。
一个进程可以被认为是工作流中的一个特定步骤，例如 RNA-seq 分析中的一个比对步骤。进程彼此独立(不需要执行任何其他进程) ，不能彼此通信/写入。数据通过输入和输出通道在进程之间传递。
例如，下面的命令行是用于创建酵母转录组索引，使用了salmon:

$ salmon index -t data/yeast/transcriptome/Saccharomyces_cerevisiae.R64-1-1.cdna.all.fa.gz -i data/yeast/salmon_index --kmerLen 31

现在我们将展示如何将其转换为一个简单的 Nextflow 进程。

进程定义

进程定义以关键字 process 开始，后跟进程名称(在本例中为 INDEX) ，最后是由花括号{}分隔的主体。进程体必须包含一个命令，或者更一般地说，表示由它执行的脚本。

process INDEX {script:"salmon index -t ${projectDir}/data/yeast/transcriptome/Saccharomyces_cerevisiae.R64-1-1.cdna.all.fa.gz -i ${projectDir}/data/yeast/salmon_index --kmerLen 31"
}

隐式变量
我们使用 Nextflow 隐式变量 ${ projectDir }来指定主脚本所在的目录。这一点很重要，因为 Nextflow 脚本是在单独的工作目录中执行的。更多的隐式变量可以参考隐式变量。

要将进程添加到工作流中，需要添加一个工作流块，并像函数一样调用该进程。我们将在工作流中了解更多的内容。
注意: 由于我们正在使用 DSL2，因此需要在脚本中包含 nextflow.enable.dsl = 2。

//process_index.nf
nextflow.enable.dsl=2process INDEX {script:"salmon index -t ${projectDir}/data/yeast/transcriptome/Saccharomyces_cerevisiae.R64-1-1.cdna.all.fa.gz -i data/yeast/salmon_index --kmerLen 31"
}workflow {//process is called like a function in the workflow blockINDEX()
}

运行脚本：nextflow run process_index.nf.

定义进程体

前面的示例是一个没有定义输入和输出的只运行一次的简单进程。为了控制输入、输出和命令的执行，一个进程可能包含五个定义块:

directives - 0，1，或多个：允许定义影响当前进程执行的可选设置，例如任务使用的 CPU 数量和分配的内存量。
**input ** - 0，1，或多个：定义输入依赖项(通常是通道) ，它决定进程执行的次数。
outputs - 0，1，或多个：定义进程用于发送结果/数据的输出通道。
when 子句 - 可选：定义为执行进程而必须验证的条件。
script - 必须：定义进程执行其任务时所执行的命令。

语法定义如下：

process < NAME > {[ directives ]        input:                < process inputs >output:               < process outputs >when:                 < condition >[script|shell|exec]:  < user script to be executed >
}

Script(脚本)

一个进程块必须包含一个脚本块。脚本块定义进程执行的命令以执行其任务。这些通常是在终端上运行的命令。
一个进程只包含一个脚本块，并且它必须是该进程包含输入和输出声明时的最后一条语句。
脚本块可以是一个简单的带引号的单行字符串。例如：

nextflow.enable.dsl=2process PROCESSBAM {script:"samtools sort -o ref1.sorted.bam ${projectDir}/data/yeast/bams/ref1.bam"
}workflow {PROCESSBAM()
}

或者，对于跨多行的命令，可以使用三重引号"""。例如：

//process_multi_line.nf
nextflow.enable.dsl=2process PROCESSBAM {script:"""samtools sort -o ref1.sorted.bam ${projectDir}/data/yeast/bams/ref1.bamsamtools index ref1.sorted.bamsamtools flagstat ref1.sorted.bam"""
}workflow {PROCESSBAM()
}

默认情况下，process 命令被解释为 Bash 脚本。然而，任何其他的脚本语言都可以简单地用相应的 Shebang 声明来启动脚本。例如:

//process_python.nf
nextflow.enable.dsl=2process PYSTUFF {script:"""#!/usr/bin/env pythonimport gzipreads = 0bases = 0with gzip.open('${projectDir}/data/yeast/reads/ref1_1.fq.gz', 'rb') as read:for id in read:seq = next(read)reads += 1bases += len(seq.strip())next(read)next(read)print("reads", reads)print("bases", bases)"""
}workflow {PYSTUFF()
}

//process_rscript.nf
nextflow.enable.dsl=2process RSTUFF {script:"""#!/usr/bin/env Rscriptlibrary("ShortRead")countFastq(dirPath="data/yeast/reads/ref1_1.fq.gz")"""
}workflow {RSTUFF()
}

允许使用不同的编程语言，这样可能更适合一个特定的工作。但是，对于大块代码，建议将它们保存到单独的文件中，并从进程脚本调用它们。

nextflow.enable.dsl=2process PYSTUFF {script:"""myscript.py"""
}workflow {PYSTUFF()
}

上面示例中的 myscript.py 脚本可以存储在与调用它们的 Nextflow 工作流脚本相同目录级别的 bin 文件夹中，并且给与执行权限。Nextflow 会自动将此文件夹添加到 PATH 环境变量

脚本参数

可以使用 Nextflow 变量(例如 ${ projectDir })动态定义脚本块中的命令。

与 bash 脚本类似，Nextflow 使用$字符来引入变量替换，要展开的变量名称可以括在大括号{ variable_name }中。

在下面的示例中，变量 kmer 被设置为值31。在脚本块中的多行字符串语句中使用 $kmer 语法引用该变量。Nextflow 变量可以在脚本块中多次使用。

//process_script.nf
nextflow.enable.dsl=2kmer = 31process INDEX {script:"""salmon index \-t $projectDir/data/yeast/transcriptome/Saccharomyces_cerevisiae.R64-1-1.cdna.all.fa.gz \-i index \--kmer $kmerecho "kmer size is $kmer""""
}workflow {INDEX()
}

在下面的示例中，我们在 Nextflow 脚本中定义了默认值为31的变量 params.kmer。

//process_script_params.nf
nextflow.enable.dsl=2params.kmer = 31process INDEX {script:"""salmon index \-t $projectDir/data/yeast/transcriptome/Saccharomyces_cerevisiae.R64-1-1.cdna.all.fa.gz \-i index \--kmer $params.kmerecho "kmer size is $params.kmer""""
}workflow {INDEX()
}

我们可以使用下面的命令运行 Nextflow 脚本，将 kmer 的默认值更改为11。

nextflow run process_script_params.nf --kmer 11

Bash 变量

Nextflow 使用相同的 Bash 语法进行变量替换，即 $variable。但是，Bash 变量需要使用转义 \$variable。
在下面的示例中，我们将 bash 变量 KMERSIZE 设置为 $params.kmer 的值，然后在脚本块中使用 KMERSIZE。

//process_escape_bash.nf
nextflow.enable.dsl=2process INDEX {script:"""#set bash variable KMERSIZEKMERSIZE=$params.kmersalmon index -t $projectDir/data/yeast/transcriptome/Saccharomyces_cerevisiae.R64-1-1.cdna.all.fa.gz -i index --kmer \$KMERSIZEecho "kmer size is $params.kmer""""
}params.kmer = 31workflow {INDEX()
}

Shell

另一种选择是使用 shell 块定义而不是脚本script。当使用 shell 语句时，通常以 $my_bash_variable 的方式引用 bash 变量。但是，shell 语句对 Nextflow 变量替换使用了不同的语法:!{nextflow_variable}。
例如，在下面使用 shell 语句的脚本中，我们将 Nextflow 变量引用为!{ projectDir }和!{ params.kmer } ，Bash 变量为 ${ KMERSIZE }。

//process_shell.nf
nextflow.enable.dsl=2params.kmer = 31process INDEX {shell:'''#set bash variable KMERSIZEKMERSIZE=!{params.kmer}salmon index -t !{projectDir}/data/yeast/transcriptome/Saccharomyces_cerevisiae.R64-1-1.cdna.all.fa.gz -i index --kmer ${KMERSIZE}echo "kmer size is !{params.kmer}"'''
}workflow {INDEX()
}

条件语句

有时，我们希望根据某些条件更改进程的运行方式。在 Nextflow 脚本中，我们可以使用条件语句，例如 if 语句。

if

If 语句使用与其他编程语言(如 Java、 C、 JavaScript 等)相同的语法。

if( < boolean expression > ) {// true branch
}
else if ( < boolean expression > ) {// true branch
}
else {// false branch
}

例如：

//process_conditional.nf
nextflow.enable.dsl=2params.aligner = 'kallisto'
params.transcriptome = "$projectDir/data/yeast/transcriptome/Saccharomyces_cerevisiae.R64-1-1.cdna.all.fa.gz"
params.kmer = 31process INDEX {script:if( params.aligner == 'kallisto' ) {"""echo indexed using kallistokallisto index -i index -k $params.kmer $params.transcriptome"""}  else if( params.aligner == 'salmon' ) {"""echo indexed using salmonsalmon index -t $params.transcriptome -i index --kmer $params.kmer"""}  else {"""echo Unknown aligner $params.aligner"""}
}workflow {INDEX()
}

输入（Inputs）

进程彼此隔离，但可以通过 Nextflow 通道从输入和输出块发送值和文件进行通信。
输入块定义进程期望从哪个通道接收输入。输入通道中元素的数量决定了进程依赖关系和进程执行的次数。

一次只能定义一个输入块，它必须包含一个或多个输入声明。
输入块语法如下：

input:<input qualifier> <input name>

<input qualifier>声明要接收的数据类型，取值类型如下：

val：通过名称作为进程脚本中的变量访问接收到的输入值。
env：使用输入值设置一个名为指定输入名称的环境变量。
path：将接收到的值作为文件处理，并在执行上下文中正确地分段处理该文件。
stdin：将接收到的值转发到进程 stdin 特殊文件。
tuple：处理具有上述限定符之一的一组输入值。
each：为输入集合中的每个条目执行流程。

//process_input_value.nf
nextflow.enable.dsl=2process PRINTCHR {input:val chrscript:"""echo processing chromosome $chr"""
}chr_ch = Channel.of( 1..22, 'X', 'Y' )workflow {PRINTCHR(chr_ch)
}

在上面的示例中，该进程执行了24次; 每次从队列通道 chr_ch 接收到一个值时，就用它来运行该进程。

通道保证按照发送的顺序交付项目，但是由于进程是以并行方式执行的，因此不能保证按照与接收项目相同的顺序处理项目。

输入文件

当需要处理文件作为输入时，需要使用path限定符。
例如，在下面的脚本中，我们使用path限定符将变量名称读取到输入文件。使用脚本块中的变量替换语法 ${ read }引用该文件:

//process_input_file.nf
nextflow.enable.dsl=2process NUMLINES {input:path readscript:"""printf '${read} 'gunzip -c ${read} | wc -l"""
}reads_ch = Channel.fromPath( 'data/yeast/reads/ref*.fq.gz' )workflow {NUMLINES(reads_ch)
}

结合输入通道

进程的一个关键特性是处理来自多个通道的输入的能力。例如：

//process_combine.nf
nextflow.enable.dsl=2process COMBINE {input:val xval yscript:"""echo $x and $y"""
}num_ch = Channel.of(1, 2, 3)
letters_ch = Channel.of('a', 'b', 'c')workflow {COMBINE(num_ch, letters_ch)
}

进程被执行3次，结果如下：

2 and b1 and a3 and c

当不是所有通道都有相同数量的元素时会发生什么？
例如：

//process_combine_02.nf
nextflow.enable.dsl=2process COMBINE {input:val xval yscript:"""echo $x and $y"""
}ch_num = Channel.of(1, 2)
ch_letters = Channel.of('a', 'b', 'c', 'd')workflow {COMBINE(ch_num, ch_letters)

在上面的示例中，进程只执行两次，因为当队列通道没有更多的数据要处理时，它将停止进程的执行。

输出：
2 and b
1 and a

与队列通道不同，值通道可以使用多次：

//process_combine_03.nf
nextflow.enable.dsl=2process COMBINE {input:val xval yscript:"""echo $x and $y"""
}
ch_num = Channel.value(1)
ch_letters = Channel.of('a', 'b', 'c')workflow {COMBINE(ch_num, ch_letters)
}

本例中进程将会执行3次：

1 and b
1 and a
1 and c

我们在前面看到，默认情况下，进程运行的次数由具有最少项的队列通道决定。each限定符允许为列表或队列通道中的每一项重复执行进程。例如：

//process_repeat.nf
nextflow.enable.dsl=2process COMBINE {input:val xeach yscript:"""echo $x and $y"""
}ch_num = Channel.of(1, 2)
ch_letters = Channel.of('a', 'b', 'c', 'd')workflow {COMBINE(ch_num, ch_letters)
}

运行$ nextflow run process_repeat.nf -process.echo。
进程将会运行8次：

2 and d
1 and a
1 and c
2 and b
2 and c
1 and d
1 and b
2 and a

总结

Nextflow 进程是工作流中的独立的步骤。
进程包含多达五个定义块，包括: 指令、输入、输出、 when 子句和最后一个脚本块。
脚本块包含要运行的命令。
一个进程应该有一个脚本，但是其他四个块是可选的。

5 Processes 上相关推荐

关于Mongodb的全面总结，学习mongodb的人，可以从这里开始！
转载地址:http://blog.csdn.net/he90227/article/details/45674513 原文地址:http://blog.csdn.NET/jakenson/articl ...
在OpenShift平台上验证NVIDIA DGX系统的分布式多节点自动驾驶AI训练
在OpenShift平台上验证NVIDIA DGX系统的分布式多节点自动驾驶AI训练自动驾驶汽车的深度神经网络(DNN)开发是一项艰巨的工作.本文验证了DGX多节点,多GPU,分布式训练在DXC机器 ...
在Ubuntu 14.04 64bit上安装MongoDB并测试
临时需要使用python来操作mongoDB, 网上搜索发现很多文档都已经非常陈旧了, 并不符合目前3.0版本的安装配置, 特此自己整理一下, 以作备忘. MongoDB简介 MongoDB(http ...
在Server 2003上部署IIS+PHP+MySQL配置清单
在Server 2003上部署IIS+PHP+MySQL I.安装Windows Server 2003 将光盘放入光驱中,设置BIOS,从CDROM引导加载安装程序,等待启动: 设置注册信息,名字和 ...
聊聊Oracle 11g的Snapshot Standby Database（上）
Oracle 11g是Data Guard的重要里程碑版本.在11g中,Active DataGuard.Advanced Compression等特性大大丰富了Data Guard的功能和在实践领域 ...
CentOS7上编译多版本PHP并同时运行及systemd设置
为什么80%的码农都做不了架构师?>>> CentOS7最大改变是systemd代替过去的systemV服务,于是配置服务的方式改变了,用systemctl替代过去的servi ...
windows 如何安装oracle 补丁包,Windows Server 2003 上安装 Oracle10g(10.2.0.1)并升级至补丁(10.2.0.4) 图解...
Windows Server 2003 上安装 Oracle10g(10.2.0.1)并升级至补丁(10.2.0.4) 图解第一部分:安装 Oracle 10.2.0.1 1.选择安装方法 2.选 ...
celery源码分析-wroker初始化分析(上)
celery源码分析本文环境python3.5.2,celery4.0.2,django1.10.x系列 celery与Django的配合使用首先,在安装有django的环境中创建一个django ...
2006年下半年网络工程师上下午试卷【附带答案】
2004-2017年真题+答案[0.23RMB]自取 2006年下半年网络工程师上午试卷 ● 若内存按字节编址,用存储容量为32K X 8 比特的存储器芯片构成地址编号A0000H至DFFFFH的 ...

5 Processes 上