2.10. 工作流#
工作流 (workflow) 指的是将命令行工具、表达式工具或（子）工作流等作为步骤进行执行的 CWL 处理单元。一个工作流必须具备 CWL 程序所定义的 inputs（输入）, outputs（输出）, 以及 step（步骤）.
CWL 工作流#
CWL 文件 echo-uppercase.cwl 定义了之前的例子中运行命令行工具的工作流以及表达式工具。
echo-uppercase.cwl#cwlVersion: v1.2
class: Workflow
requirements:
InlineJavascriptRequirement: {}
inputs:
message: string
outputs:
out:
type: string
outputSource: uppercase/uppercase_message
steps:
echo:
run: echo.cwl
in:
message: message
out: [out]
uppercase:
run: uppercase.cwl
in:
message:
source: echo/out
out: [uppercase_message]
命令行工具或表达式工具亦可直接写在定义工作流的单一 CWL 文件里。例如，我们可以将 echo-uppercase.cwl 改写成一个单独的文件如下：
echo-uppercase-single-file.cwl#cwlVersion: v1.2
class: Workflow
requirements:
InlineJavascriptRequirement: {}
inputs:
message: string
outputs:
out:
type: string
outputSource: uppercase/uppercase_message
steps:
echo:
run:
class: CommandLineTool
baseCommand: echo
stdout: output.txt
inputs:
message:
type: string
inputBinding: {}
outputs:
out:
type: string
outputBinding:
glob: output.txt
loadContents: true
outputEval: $(self[0].contents)
in:
message: message
out: [out]
uppercase:
run:
class: ExpressionTool
requirements:
InlineJavascriptRequirement: {}
inputs:
message: string
outputs:
uppercase_message: string
expression: |
${ return {"uppercase_message": inputs.message.toUpperCase()}; }
in:
message:
source: echo/out
out: [uppercase_message]
将代码分为多个独立的文件有助于模块化和代码的组织条理。不过，把全部代码集中在一个文件，可能对开发更有利。将多个文件整合为一个，还有其他手段（如 cwltool --pack），将在本《指南》的其他章节进一步讨论。
备注
子工作流 (sub-workflow) 需要启用 SubworkflowFeatureRequirement 这一要求，这在另一章节中将有更详细的教程。
2.10.1. 编写工作流#
这个工作流从 tar 文件中提取一个 Java 源文件，然后编译。
1st-workflow.cwl##!/usr/bin/env cwl-runner
cwlVersion: v1.2
class: Workflow
inputs:
tarball: File
name_of_file_to_extract: string
outputs:
compiled_class:
type: File
outputSource: compile/classfile
steps:
untar:
run: tar-param.cwl
in:
tarfile: tarball
extractfile: name_of_file_to_extract
out: [extracted_file]
compile:
run: arguments.cwl
in:
src: untar/extracted_file
out: [classfile]
1st-workflow.cwl 的图示
在分立的文件中通过 YAML 或 JSON 对象来描述一次运行的输入：
1st-workflow-job.yml#tarball:
class: File
path: hello.tar
name_of_file_to_extract: Hello.java
接下来，创建一个 Java 文件样本，将其打包到一个 tar 文件，以供运行命令行工具时使用。
$ echo "public class Hello {}" > Hello.java && tar -cvf hello.tar Hello.java
Hello.java
现在，在命令行上以工具描述和输入对象为参数调用 cwltool:
$ cwltool 1st-workflow.cwl 1st-workflow-job.yml
INFO /home/docs/checkouts/readthedocs.org/user_builds/common-workflow-languageuser-guide-zh-hans/envs/latest/bin/cwltool 3.2.20260411152607
INFO Resolved '1st-workflow.cwl' to 'file:///home/docs/checkouts/readthedocs.org/user_builds/common-workflow-languageuser-guide-zh-hans/checkouts/latest/src/_includes/cwl/workflows/1st-workflow.cwl'
INFO [workflow ] start
INFO [workflow ] starting step untar
INFO [step untar] start
INFO [job untar] /tmp/pyjenkc6$ tar \
--extract \
--file \
/tmp/6tx216_5/stg2e882b12-cb68-4c50-95c1-9ad9f14d4c5d/hello.tar \
Hello.java
INFO [job untar] completed success
INFO [step untar] completed success
INFO [workflow ] starting step compile
INFO [step compile] start
ERROR Workflow error, try again with --debug for more information:
Docker is not available for this tool, try --no-container to disable Docker, or install a user space Docker replacement like uDocker with --user-space-docker-cmd.: docker executable is not available
这是怎么一回事呢？我们一一道来：
cwlVersion: v1.0
class: Workflow
cwlVersion 字段指明该文件使用的 CWL 规约版本。class 字段表明该文件描述的是一个工作流。
inputs:
tarball: File
name_of_file_to_extract: string
inputs 代码段描述的是工作流的输入，即一组输入参数。这里面的每个参数都由标识符和数据类型构成，它们可以用作工作流中某个指定步骤的输入源。
outputs:
compiled_class:
type: File
outputSource: compile/classfile
outputs 代码段描述的是工作流的输出。这同样是一组由标识符和数据类型构成的参数。outputSource 将 compile（编译）步骤的输出参数 classfile 同工作流的输出参数 compiled_class 建立联系。
steps:
untar:
run: tar-param.cwl
in:
tarfile: tarball
extractfile: name_of_file_to_extract
out: [extracted_file]
steps 代码段描述的是工作流的实际步骤。这个例子中，第一个步骤是从 tar 归档文件中提取一个文件，而第二步是使用 Java 编译器编译来自第一步的文件。工作流的各个步骤不是必须按照在代码中列出的顺序运行，而是由各个步骤之间（由 source 决定）的依赖关系确定其先后次序。此外，工作流中没有依赖关系的多个步骤允许并行运行。
第一个步骤 untar 运行 tar-param.cwl（此前见于《参数引用》一节）。该工具有两个输入参数 tarfile 和 extractfile, 以及一个输出参数 extracted_file.
该工作流步骤下的 in 这段代码将上述两个输入参数同工作流的输入 tarball 和 name_of_file_to_extract 通过source 关联起来。这意味着当这个工作流步骤执行时，为了运行其指定的工具，tarball 和 name_of_file_to_extract 所赋予的值将用于 tarfile 和 extractfile 参数。
工作流步骤下的 out 代码段列出了预期中从工具应获得的参数。
compile:
run: arguments.cwl
in:
src: untar/extracted_file
out: [classfile]
第二个步骤 compile（编译）依赖于第一步的结果，具体而言它的输入参数 src 与 untar 步骤的输出参数 untar/extracted_file 相关联。这一步骤运行 arguments.cwl（此前见于《附加参数》一节）。此步骤的输出 classfile 关联的是工作流的 outputs 部分（见上）。
2.10.2. 嵌套的工作流#
工作流的作用在于将多种工具组合起来，进行更大规模的操作。我们还可以将一个工作流整体视为一个工具；如果工作流引擎支持 SubworkflowFeatureRequirement, 则 CWL 工作流可以用作另一 CWL 工作流中的单个步骤：
requirements:
SubworkflowFeatureRequirement: {}
下面这个例子里的工作流嵌入了我们的 1st-workflow.cwl 工作流：
nestedworkflows.cwl##!/usr/bin/env cwl-runner
cwlVersion: v1.2
class: Workflow
inputs: []
outputs:
classout:
type: File
outputSource: compile/compiled_class
requirements:
SubworkflowFeatureRequirement: {}
steps:
compile:
run: 1st-workflow.cwl
in:
tarball: create-tar/tar_compressed_java_file
name_of_file_to_extract:
default: "Hello.java"
out: [compiled_class]
create-tar:
in: []
out: [tar_compressed_java_file]
run:
class: CommandLineTool
requirements:
InitialWorkDirRequirement:
listing:
- entryname: Hello.java
entry: |
public class Hello {
public static void main(String[] argv) {
System.out.println("Hello from Java");
}
}
inputs: []
baseCommand: [tar, --create, --file=hello.tar, Hello.java]
outputs:
tar_compressed_java_file:
type: File
streamable: true
outputBinding:
glob: "hello.tar"
备注
Visualization of the workflow and the inner workflow from its `compile` step
这个由两个步骤构成的工作流，起始于 create-tar 步骤，进而接入橙色的 compile 步骤；如右侧图所示，compile 是另一个工作流。由紫色的部分可见字符串常量 "Hello.java" 赋值给 name_of_file_to_extract.
CWL Workflow（工作流）就和 CommandLineTool（命令行工具）一样，可以充当一个步骤。通过 run 字段可以将其 CWL 文件包含进来。然后，工作流的输入 (tarball 和 name_of_file_to_extract) 与输出 (compiled_class) 就可以映射为这一步骤的输入/输出。
compile:
run: 1st-workflow.cwl
in:
tarball: create-tar/tar_compressed_java_file
name_of_file_to_extract:
default: "Hello.java"
out: [compiled_class]
我们的 1st-workflow.cwl 文件基于工作流输入进行了参数化，因此运行时必须提供作业文件 (job file) 以指明 tar 文件和 *.java 的文件名。一般而言这样做是最合适的，好处在于多个上级工作流、乃至同一工作流中的多个步骤间可以重复使用代码。
这里，我们用 default: 字段，将输入 name_of_file_to_extract 硬性编码为 "Hello.java", 然而我们的工作流仍然需要 tarball 字段指定的一个 tar 文件，我们会通过 create-tar 这个步骤将它创建好。到这一步，比较好的办法是重整 1st-workflow.cwl 中的代码，像用作工具时一样使用更具体的输入/输出名。
另一种可能是采用更加特殊化的办法，避免作业文件中的外部依赖。在这个工作流中， 我们可以利用前述 InitialWorkDirRequirement 需求，生成一个硬性编码的 Hello.java 文件，然后将它添加到 tar 文件。
create-tar:
requirements:
InitialWorkDirRequirement:
listing:
- entryname: Hello.java
entry: |
public class Hello {
public static void main(String[] argv) {
System.out.println("Hello from Java");
}
}
这种情况下，我们的步骤中可以直接以 Hello.java 为前提假定，无需参数化，从而在 baseCommand 以及产出的 outputs 中使用硬性编码的值 hello.tar 和 Hello.java:
run:
class: CommandLineTool
inputs: []
baseCommand: [tar, --create, --file=hello.tar, Hello.java]
outputs:
tar_compressed_java_file:
type: File
streamable: true
outputBinding:
glob: "hello.tar"
您是否注意到，我们并没有将 tar --create 工具划分到一个独立的文件，而是嵌入到 CWL 工作流文件中？一般而言这样做并不是最合适的，将导致工具无法重复使用。在这个特例中，如此操作的原因是命令行已是硬性编码的，其中的文件名仅对这一个工作流有效。
这个例子里我们不得不在外部创制 tar 文件，但这无非是因为我们将内部工作流设计成以该文件为输入。内部工作流更好的重整方式，可以是令其接受一个待编译 Java 文件的列表，使之作为工具步骤用于其它工作流中时的用法得以简化。
嵌套的工作流作为一种强大的功能，可以用于生成高阶函数和可复用的工作流单元——然而，正如创建 CWL 工具描述时一样，我们必须用心于提高它在多个工作流间的可用性。
2.10.3. 分散步骤#
了解了如何编写工作流，我们可以准备使用 ScatterFeatureRequirement. 该功能用于告诉运行程序，您打算对列表中的多个输入重复运行某个工具或者工作流。这样，工作流可以将输入作为数组，并对每个数组元素运行指定的步骤，如同单个输入一样。这样， 我们可以让同一工作流对多个输入运行，而无需生成多个不同的命令或者 YAML 输入文件。
requirements:
ScatterFeatureRequirement: {}
用户第一次接触分散 (scatter) 功能，最主要的原因一般是为了对多个不同的样本进行同一分析。我们从一个简单的工作流出发，调用我们的第一个示例 (hello_world.cwl), 并以一个字符串数组作为工作流的输入：
scatter-workflow.cwl##!/usr/bin/env cwl-runner
cwlVersion: v1.2
class: Workflow
requirements:
ScatterFeatureRequirement: {}
inputs:
message_array: string[]
steps:
echo:
run: hello_world.cwl
scatter: message
in:
message: message_array
out: []
outputs: []
除了 requirements 代码段加入了 ScatterFeatureRequirement 以外，还有什么新情况？
inputs:
message_array: string[]
首先，请注意，主工作流级别的输入这时需要字符串数组。
steps:
echo:
run: hello_world.cwl
scatter: message
in:
message: message_array
out: []
这里，我们在 echo 步骤下添加了名为 scatter 的新字段。该字段告诉运行程序，我们希望让这一特定步骤对输入分散运行。请注意，"scatter" 后列出的输入名是该步骤的输入之一，而非工作流层面的输入。
我们第一次做分散，就是这么简单。因为我们的工具并不收集任何输出，我们仍然可以在工作流中写上 outputs: []. 但是，如果你知道工作流的最终输出需要归集多个输出，请记得这里同样要更改为数组类型！
使用如下输入文件：
scatter-job.yml#message_array:
- Hello world!
- Hola mundo!
- Bonjour le monde!
- Hallo welt!
提醒一下，hello_world.cwl 仅仅是对某段文字调用 echo 这个命令。如果我们在命令行调用 cwltool scatter-workflow.cwl scatter-job.yml:
$ cwltool scatter-workflow.cwl scatter-job.yml
INFO /home/docs/checkouts/readthedocs.org/user_builds/common-workflow-languageuser-guide-zh-hans/envs/latest/bin/cwltool 3.2.20260411152607
INFO Resolved 'scatter-workflow.cwl' to 'file:///home/docs/checkouts/readthedocs.org/user_builds/common-workflow-languageuser-guide-zh-hans/checkouts/latest/src/_includes/cwl/workflows/scatter-workflow.cwl'
INFO [workflow ] start
INFO [workflow ] starting step echo
INFO [step echo] start
INFO [job echo] /tmp/tzh_ryfw$ echo \
'Hello world!' > /tmp/tzh_ryfw/f2ee667fba65d968d4f6092765fa965e792bb20b
INFO [job echo] completed success
INFO [step echo] start
INFO [job echo_2] /tmp/m0op7tfm$ echo \
'Hola mundo!' > /tmp/m0op7tfm/f2ee667fba65d968d4f6092765fa965e792bb20b
INFO [job echo_2] completed success
INFO [step echo] start
INFO [job echo_3] /tmp/won_ssmw$ echo \
'Bonjour le monde!' > /tmp/won_ssmw/f2ee667fba65d968d4f6092765fa965e792bb20b
INFO [job echo_3] completed success
INFO [step echo] start
INFO [job echo_4] /tmp/jeo99p_u$ echo \
'Hallo welt!' > /tmp/jeo99p_u/f2ee667fba65d968d4f6092765fa965e792bb20b
INFO [job echo_4] completed success
INFO [step echo] completed success
INFO [workflow ] completed success
{}INFO Final process status is success
可以看到，这个工作流分多次对 message_array 数组的每个元素一一调用 "echo". 好的，那么如果我们想要让工作流中两个步骤进行分散，该怎么办？
和之前一样，我们来做一个简单的回显 (echo), 但这次将捕获 stdout（标准输出）。为此，我们用以下数行代码取代 outputs: []
hello_world_to_stdout.cwl#outputs:
echo_out:
type: stdout
进而添加第二个步骤，用 wc 命令计算每个文件中的字符数量。请见如下工具：
wc-tool.cwl##!/usr/bin/env cwl-runner
cwlVersion: v1.2
class: CommandLineTool
baseCommand: wc
arguments: ["-c"]
inputs:
input_file:
type: File
inputBinding:
position: 1
outputs: []
现在，该如何加入分散操作呢？请记住，"scatter"（分散）字段出现在每个步骤下：
scatter-two-steps.cwl##!/usr/bin/env cwl-runner
cwlVersion: v1.2
class: Workflow
requirements:
ScatterFeatureRequirement: {}
inputs:
message_array: string[]
steps:
echo:
run: hello_world_to_stdout.cwl
scatter: message
in:
message: message_array
out: [echo_out]
wc:
run: wc-tool.cwl
scatter: input_file
in:
input_file: echo/echo_out
out: []
outputs: []
这里，我们在每一步骤下加入了 "scatter" 字段。对这个具体的示例而言这样做无妨，因为这段程序运行得很快。但是，如果要对较多样本运行更复杂的工作流，您可能得另寻途径。这里，我们在每个步骤中独立地进行分散，但既然第二个步骤并不实际依赖第一个步骤对所有语种（即输入数组中的一个元素）全部完成操作，我们并没有高效地运用分散功能。第二个步骤期待从第一个步骤得来的数组作为其输入，因此将一直等待到第一步骤完全结束才会开始运行。打个比方，假设运行 echo Hello World! 需要花1分钟，对其输出运行 wc -c 需要3分钟，然后 echo Hallo welt! 需要5分钟来运行，最后对其输出运行 wc 需要3分钟。即使 echo Hello World! 本可以在4分钟内完成，它也要实际花费8分钟，因为第一个步骤必须等待 echo Hallo welt! 完成。很显然，运算量大时这样的表现会很差劲。
好的，那么，我们对不依赖其它样本、可以独立进行的步骤，该如何分散？请回想嵌套的工作流部分，其中写道，我们可以让整个工作流称为另一个工作流中的一个步骤！我们将两步骤的工作流改写为单一步骤的子工作流：
scatter-nested-workflow.cwl##!/usr/bin/env cwl-runner
cwlVersion: v1.2
class: Workflow
requirements:
ScatterFeatureRequirement: {}
SubworkflowFeatureRequirement: {}
inputs:
message_array: string[]
steps:
subworkflow:
run:
class: Workflow
inputs:
message: string
outputs: []
steps:
echo:
run: hello_world_to_stdout.cwl
in:
message: message
out: [echo_out]
wc:
run: wc-tool.cwl
in:
input_file: echo/echo_out
out: []
scatter: message
in:
message: message_array
out: []
outputs: []
这样，分散就可以针对一个单独的步骤进行，不过该步骤本身分为两个子步骤。这就让每个分散的步骤并行进行。
2.10.4. 条件性工作流#
这个工作流包含一个取决于输入的有条件步骤，可以根据程序起始或先前步骤给出的输入参数决定是否跳过某些额外步骤。
conditional-workflow.cwl#class: Workflow
cwlVersion: v1.2
inputs:
val: int
steps:
step1:
in:
in1: val
a_new_var: val
run: foo.cwl
when: $(inputs.in1 < 1)
out: [out1]
step2:
in:
in1: val
a_new_var: val
run: foo.cwl
when: $(inputs.a_new_var > 2)
out: [out1]
outputs:
out1:
type: string
outputSource:
- step1/out1
- step2/out1
pickValue: first_non_null
requirements:
InlineJavascriptRequirement: {}
MultipleInputFeatureRequirement: {}
首先要注意，这个工作流只能兼容于版本1.2或以上的 CWL 标准。
class: Workflow
cwlVersion: v1.2
工作流的第一个步骤 (step1) 包含两个输入字段，它将在条件满足时执行 foo.cwl. 这里新出现的字段 when 就是进行条件检验的地方。这个例子里，只有当 in1 从工作流获取的值小于1 (< 1) 时，此步骤才会执行。
steps:
step1:
in:
in1: val
a_new_var: val
run: foo.cwl
when: $(inputs.in1 < 1)
out: [out1]
使用命令如 cwltool cond-wf-003.1.cwl --val 0, 则输入参数值 (0) 将满足第一个步骤的条件，从而使之得到执行，并且在日志中以 INFO [step step1] start 出现。相反，第二个步骤则跳过了，即日志中 INFO [step step2] will be skipped 所记。
INFO [workflow ] start
INFO [workflow ] starting step step1
INFO [step step1] start
INFO [job step1] /private/tmp/docker_tmpdcyoto2d$ echo
INFO [job step1] completed success
INFO [step step1] completed success
INFO [workflow ] starting step step2
INFO [step step2] will be skipped
INFO [step step2] completed skipped
INFO [workflow ] completed success
{
"out1": "foo 0"
}
INFO Final process status is success
当参数值为3, 如命令 cwltool cond-wf-003.1.cwl --val 3 所给出，则第一个条件行步骤不会执行，反之第二个会。
INFO [workflow ] start
INFO [workflow ] starting step step1
INFO [step step1] will be skipped
INFO [step step1] completed skipped
INFO [workflow ] starting step step2
INFO [step step2] start
INFO [job step2] /private/tmp/docker_tmpqwr93mxx$ echo
INFO [job step2] completed success
INFO [step step2] completed success
INFO [workflow ] completed success
{
"out1": "foo 3"
}
INFO Final process status is success
如果条件都不成立，如用 --val 2 所导致的，则工作流将发起 permanentFail 异常。
$ cwltool cond-wf-003.1.cwl --val 2
INFO [workflow ] start
INFO [workflow ] starting step step1
INFO [step step1] will be skipped
INFO [step step1] completed skipped
INFO [workflow ] starting step step2
INFO [step step2] will be skipped
INFO [step step2] completed skipped
ERROR [workflow ] Cannot collect workflow output: All sources for 'out1' are null
INFO [workflow ] completed permanentFail
WARNING Final process status is permanentFail