全部产品
云市场

Cromwell 工作流引擎支持

更新时间:2019-08-30 10:34:24

Cromwell 是 Broad Institute 开发的工作流管理系统,当前已获得阿里云批量计算服务的支持。通过 Cromwell 可以将 WDL 描述的 workflow 转化为批量计算的作业(Job)运行。用户将为作业运行时实际消耗的计算和存储资源付费,不需要支付资源之外的附加费用。本文将介绍如何使用 Cromwell 在阿里云批量计算服务上运行工作流。

1. 准备工作

A) 开通批量计算服务

要使用批量计算服务,请根据官方文档里面的指导开通批量计算和其依赖的相关服务,如OSS等。

注意:创建 OSS Bucket 的区域,需要和使用批量计算的区域一致。

B) 下载 Cromwell

下载地址

C) 开通 ECS 作为 Cromwell server

当前批量计算提供了 Cromwell server 的 ECS 镜像,用户可以用此镜像开通一台 ESC 作为 server。镜像中提供了 Cromwell 官网要求的基本配置和常用软件。在此镜像中,Cromwell 的工作目录位于/home/cromwell,上一步下载的 Crowwell jar 包可以放置在 /home/cromwell/cromwell 目录下。

注意:用户也可以自己按照 Cromwell 官方的要求自己搭建 Cromwell server, 上面的镜像只是提供了方便的方式,不是强制要求。

2. 使用 Cromwell

配置文件

Cromwell 运行的配置文件,包括:

  • Cromwell 公共配置。
  • 批量计算相关配置,包含了批量计算作为后端需要的存储、计算等资源配置。

关于配置参数的详细介绍请参考 Cromwell 官方文档。如下是一个批量计算配置文件的例子 bcs.conf:

  1. include required(classpath(application))
  2. # Cromwell system settings
  3. system {
  4. # Cromwell will cap the number of running workflows at N
  5. max-concurrent-workflows = 1000
  6. }
  7. docker {
  8. hash-lookup {
  9. enabled = false
  10. }
  11. }
  12. database {
  13. profile = slick.jdbc.MySQLProfile$
  14. db {
  15. driver = com.mysql.jdbc.Driver
  16. url = jdbc:mysql://localhost/db_cromwell?rewriteBatchedStatements=true&useSSL=false
  17. user = user_cromwell
  18. password = "xxxxxx"
  19. connectionTimeout = 5000
  20. }
  21. }
  22. engine {
  23. filesystems {
  24. oss {
  25. auth {
  26. endpoint = "oss-cn-shanghai.aliyuncs.com"
  27. access-id = "xxxx"
  28. access-key = "yyyy"
  29. }
  30. }
  31. }
  32. }
  33. backend {
  34. default = BCS
  35. providers {
  36. BCS {
  37. actor-factory = cromwell.backend.impl.bcs.BcsBackendLifecycleActorFactory
  38. config {
  39. root = oss://your-bucket/cromwell_dir
  40. region = cn-shanghai
  41. access-id = "xxxx"
  42. access-key = "xxxx"
  43. filesystems {
  44. oss {
  45. auth {
  46. endpoint = oss-cn-shanghai.aliyuncs.com
  47. access-id = "xxxx"
  48. access-key = "xxxx"
  49. }
  50. }
  51. }
  52. default-runtime-attributes {
  53. failOnStderr: false
  54. continueOnReturnCode: 0
  55. cluster: OnDemand ecs.sn1.medium img-ubuntu-vpc
  56. vpc: 192.168.0.0/16
  57. autoReleaseJob: false
  58. }
  59. }
  60. }
  61. }
  62. }

如果使用前面章节中的镜像开通 ECS 作为 Cromwell server,配置文件位于 /home/cromwell/cromwell/bcs_sample.conf,只需要填写自己的配置即可使用 Cromwell。

注意:Cromwell 可以在公网环境(如本地服务器、配置了公网 IP 的阿里云 ECS 等)运行,也可以在阿里云 VPC 环境下运行。在 VPC 环境下使用时,有如下几处要修改为 VPC 内网下的配置:

  • OSS 的内网 endpoint :
    • engine.filesystems.oss.auth.endpoint = "oss-cn-shanghai-internal.aliyuncs.com"
    • backend.providers.BCS.config.filesystems.oss.auth.endpoint = "oss-cn-shanghai-internal.aliyuncs.com"
  • 添加批量计算的内网 endpoint:
    • backend.providers.BCS.config.user-defined-region = "cn-shanghai-vpc"
    • backend.providers.BCS.config.user-defined-domain = "batchcompute-vpc.cn-shanghai.aliyuncs.com"

运行模式

Cromwell支持两种模式

  • run 模式
  • server 模式

关于两种模式的详细描述,请参考 Cromwell 官网文档。下面重点介绍这两种模式下如何使用批量计算。

A) run模式

run模式适用于本地运行一个单独的 WDL 文件描述的工作流,命令行如下:java -Dconfig.file=bcs.conf -jar cromwell.jar run echo.wdl --inputs echo.inputs

  • WDL 文件:描述详细的工作流。工作流中每个 task 对应批量计算的一个作业(Job)。
  • inputs文件:是 WDL 中定义的工作流的输入信息inputs 文件是用来描述 WDL 文件中定义的工作流及其 task 的输入文件。如下所示:
    1. {
    2. workflow_name.task_name.input1: xxxxxx
    3. }

运行成功后,WDL 文件中描述的工作流中的一个 task 会作为批量计算的一个作业(Job)来提交。此时登录批量计算的控制台就可以看到当前的 Job 状态。

show_bcs_job

当 workflow 中所有的 task 对应的作业运行完成后,工作流运行完成。

B) server 模式

启动 server

相比 run 模式一次运行只能处理一个 WDL 文件,server 模式可以并行处理多个 WDL 文件。关于 server 模式的更多信息,请参考 Cromwell 官方文档。可以采用如下命令行启动 server:java -Dconfig.file=bsc.conf -jar cromwell.jar serverserver 启动成功后,就可以接收来自 client 的工作流处理请求。下面分别介绍如何使用 API 和 CLI 的方式向 server 提交工作流。

使用 API 提交工作流

server 启动后,可以通过浏览器访问 Cromwell Server,比如 Server 的 IP 为39.105.xxx.yyy,则在浏览器中输入http://39.105.xxx.yyy:8000,通过如下图所示的界面提交任务:cromwell_server更多API接口及用法,请参考 Cromwell 官网文档

使用 CLI 提交工作流[推荐]

除了可以使用 API 提交工作流以外,Cromwell 官方还提供了一个开源的 CLI 命令行工具 widder。可以使用如下的命令提交一个工作流:

  1. python widdler.py run echo.wdl echo.inputs -o bcs_workflow_tag:tagxxx -S localhost

其中-o key:value是用于设置option,批量计算提供了 bcs_workflow_tag:tagxxx 选项,用于配置作业输出目录的tag(下一节查看运行结果中会介绍)。

如果使用前面章节中的镜像开通 ECS 作为 Cromwell server,镜像中已经安装了 widdler,位于 /home/cromwell/widdler。可以使用如下的命令提交工作流:

  1. widdler run echo.wdl echo.inputs -o bcs_workflow_tag:tagxxx -S localhost

更多命令用法可使用widdler -h命令查看,或参考官方文档

3. 查看运行结果

工作流运行结束后,输出结果被上传到了配置文件或 WDL 中定义的 OSS 路径下。在OSS路径上面的目录结构如下:

cromwell_output_dir如上图所示,在配置文件中的config.root目录下有如下输出目录:

  • 第一层:workflowname 工作流的名称
  • 第二层:通过上一节中 CLI 命令的-o设置的目录tag
  • 第三层:workflow id,每次运行会生成一个
  • 第四层:workflow 中每个 task 的运行输出,比如上图中的 workflow 15e45adf-6dc7-4727-850c-89545faf81b0 有两个 task,每个task对应的目录命名是call-taskname,目录中包含三部分内容:
    • 批量计算的日志,包括 bcs-stdout 和 bcs-stderr
    • 当前 task 的输出,比如图中的 output1/output2 等
    • 当前 task 执行的 stdout 和 stderr

4. 使用建议

在使用过程中,关于 BCS 的配置,有如下的建议供参考:

使用集群

批量计算提供了两种使用集群的方式:

  • 自动集群
  • 固定集群

A) 自动集群

在config配置文件中指定默认的资源类型、实例类型以及镜像类型,在提交批量计算 Job 时就会使用这些配置自动创建集群,比如:

  1. default-runtime-attributes {
  2. cluster : "OnDemand ecs.sn1ne.large img-ubuntu-vpc"
  3. }

如果在某些 workflow 中不使用默认集群配置,也可以通过inputs文件中指定 workflow 中某个 task 的对应的批量计算的集群配置(将 cluster_config 作为 task 的一个输入),比如:

  1. {
  2. workflow_name.task_name.cluster_config: "OnDemand ecs.sn2ne.8xlarge img-ubuntu-vpc"
  3. }

然后在 task 中重新设置运行配置:

  1. task task_demo {
  2. String cluster_config
  3. runtime {
  4. cluster: cluster_config
  5. }
  6. }

就会覆盖默认配置,使用新的配置信息创建集群。

B) 固定集群

使用自动集群时,需要创建新集群,会有一个等待集群的时间。如果对于启动时间有要求,或者有了大量的作业提交,可以考虑使用固定集群。比如:

  1. default-runtime-attributes {
  2. cluster : "cls-xxxxxxxxxx"
  3. }

注意:使用固定集群时,如果使用完毕,请及时释放集群,否则集群中的实例会持续收费。

Cromwell Server 配置建议

  • 大压力作业时,建议使用较高配置的机器作为 Cromwell Server,比如ecs.sn1ne.8xlarge等32核64GB的机器。
  • 大压力作业时,修改 Cromwell Server 的最大打开文件数。比如在ubuntu下可以通过修改/etc/security/limits.conf文件,比如修改最大文件数为100万:
    1. root soft nofile 1000000
    2. root hard nofile 1000000
    3. * soft nofile 1000000
    4. * hard nofile 1000000
  • 确认 Cromwell Server 有配置数据库,防止作业信息丢失。
  • 设置 bcs.conf 里面的并发作业数,比如 system.max-concurrent-workflows = 1000

开通批量计算相关配额

如果有大压力场景,可能需要联系批量计算服务开通对应的配额,比如:

  • 一个用户所有作业的数量(包括完成的、运行的、等待的等多种状态下);
  • 同时运行的作业的集群的数量(包括固定集群和自动集群);

使用 NAS

使用 NAS 时要注意以下几点:

  • NAS 必须在 VPC 内使用,要求添加挂载点时,必须指定 VPC;
  • 所以要求在 runtime 中必须包含:
    • VPC 信息
    • mounts 信息

下面的例子可供参考:

  1. runtime {
  2. cluster: cluster_config
  3. mounts: "nas://1f****04-xkv88.cn-beijing.nas.aliyuncs.com:/ /mnt/ true"
  4. vpc: "192.168.0.0/16 vpc-2zexxxxxxxx1hxirm"
  5. }

高级特性支持

Glob

Cromwell 支持使用 glob 来指定工作流中多个文件作为 task 的输出,比如:

  1. task globber {
  2. command <<<
  3. for i in `seq 1 5`
  4. do
  5. mkdir out-$i
  6. echo globbing is my number $i best hobby out-$i/$i.txt
  7. done
  8. >>>
  9. output {
  10. Array[File] outFiles = glob("out-*/*.txt")
  11. }
  12. }
  13. workflow test {
  14. call globber
  15. }

当 task 执行结束时,通过 glob 指定的多个文件会作为输出,上传到 OSS 上。

Call Caching

Call Caching 是 Cromwell 提供的高级特性,如果检测到工作流中某个 task (对应一个批量计算的 job )和之前已经执行过的某个 task 具有相同的输入和运行时等条件,则不需要再执行,直接取之前的运行结果,这样可以为客户节省时间和费用。一个常见的场景是如果一个工作流有 n 个 task,当执行到中间某一个 task 时由于某些原因失败了,排除了错误之后,再次提交这个工作流运行后,Cromwell 判断如果满足条件,则已经完成的几个 task 不需要重新执行,只需要从出错的 task 开始继续运行。

配置 Call Caching

要在 BCS 后端情况下使用 Call Caching 特性,需要如下配置项:

  1. database {
  2. profile = slick.jdbc.MySQLProfile$
  3. db {
  4. driver = com.mysql.jdbc.Driver
  5. url = jdbc:mysql://localhost/db_cromwell?rewriteBatchedStatements=true&useSSL=false
  6. user = user_cromwell
  7. password = "xxxxx"
  8. connectionTimeout = 5000
  9. }
  10. }
  11. call-caching {
  12. # Allows re-use of existing results for jobs you have already run
  13. # (default: false)
  14. enabled = true
  15. # Whether to invalidate a cache result forever if we cannot reuse them. Disable this if you expect some cache copies
  16. # to fail for external reasons which should not invalidate the cache (e.g. auth differences between users):
  17. # (default: true)
  18. invalidate-bad-cache-results = true
  19. }
  20. docker {
  21. hash-lookup {
  22. enabled = true
  23. # How should docker hashes be looked up. Possible values are local and remote
  24. # local: Lookup hashes on the local docker daemon using the cli
  25. # remote: Lookup hashes on alibab cloud Container Registry
  26. method = remote
  27. alibabacloudcr {
  28. num-threads = 5
  29. auth {
  30. access-id = "xxxx"
  31. access-key = "yyyy"
  32. }
  33. }
  34. }
  35. }
  36. engine {
  37. filesystems {
  38. oss {
  39. auth {
  40. endpoint = "oss-cn-shanghai.aliyuncs.com"
  41. access-id = "xxxx"
  42. access-key = "yyyy"
  43. }
  44. }
  45. }
  46. }
  47. backend {
  48. default = BCS
  49. providers {
  50. BCS {
  51. actor-factory = cromwell.backend.impl.bcs.BcsBackendLifecycleActorFactory
  52. config {
  53. #其他配置省略
  54. filesystems {
  55. oss {
  56. auth {
  57. endpoint = oss-cn-shanghai.aliyuncs.com
  58. access-id = "xxxx"
  59. access-key = "yyyy"
  60. }
  61. caching {
  62. # When a cache hit is found, the following duplication strategy will be followed to use the cached outputs
  63. # Possible values: copy, reference. Defaults to copy
  64. # copy: Copy the output files
  65. # reference: DO NOT copy the output files but point to the original output files instead.
  66. # Will still make sure than all the original output files exist and are accessible before
  67. # going forward with the cache hit.
  68. duplication-strategy = reference
  69. }
  70. }
  71. }
  72. default-runtime-attributes {
  73. failOnStderr: false
  74. continueOnReturnCode: 0
  75. cluster: "OnDemand ecs.sn1.medium img-ubuntu-vpc"
  76. vpc: "192.168.0.0/16"
  77. }
  78. }
  79. }
  80. }
  81. }
  • database 配置:Cromwell 将 workflow 的执行元数据存储在数据库中,所以需要添加数据库配置,详细情况参考Cromwell 官网指导
  • call-caching 配置:Call Caching 的开关配置等;
  • docker.hash-lookup 配置: 设置 Hash 查找开关及阿里云 CR 等信息,用于查找镜像的 Hash 值。
  • backend.providers.BCS.config.filesystems.oss.caching 配置:设置 Call Caching命中后,使用原来输出的方式,批量计算在这里支持 reference 模式,不需要拷贝原有的结果,节省时间和成本。

命中条件

使用批量计算作为后端时,Cromwell 通过如下条件判断一个 task 是否需要重新执行:

条件 解释
continueOnReturnCode 公共运行时参数,可以继续执行的返回码
docker 公共运行时参数,后端的Docker配置
failOnStderr 公共运行时参数,stderr非空时是否失败
cluster 批量计算后端运行时参数,标识作业运行的集群信息
userData 批量计算后端,用户自定义数据

除了上述运行时参数外,如果 task 的输入参数也一样,Cromwell 会判定为不需要执行的 task,直接获取上次执行的结果,并继续工作流的执行。