Task Definition

conductor维护工作任务类型的注册表。必须在工作流中使用之前注册任务类型。

例如：

{

  "name": "encode_task",

  "retryCount": 3,

  "timeoutSeconds": 1200,

  "inputKeys": [

    "sourceRequestId",

    "qcElementType"

  ],

  "outputKeys": [

    "state",

    "skipped",

    "result"

  ],

  "timeoutPolicy": "TIME_OUT_WF",

  "retryLogic": "FIXED",

  "retryDelaySeconds": 600,

  "responseTimeoutSeconds": 3600

}



name： 任务类型，唯一
retryCount： 任务标记为falure时重新尝试的次数
retryLogic：重试机制
timeoutSeconds：时间（以毫秒为单位），之后任务被标记为TIMED_OUT，如果在转入IN_PROGRESS状态后未完成，设为0就没有timeout
timeoutPolicy：任务超时策略
responseTimeoutSeconds：如果大于0，则如果在此时间后未更新状态，则重新调度任务。 当conductor因为错误/网络故障而轮询任务但无法完成时很有用。
outputKeys：任务输出的一组键。 用于记录任务的输出

Retry Logic

FIXED : 在 retryDelaySeconds之后重新调度任务
EXPONENTIAL_BACKOFF : retryDelaySeconds * attempNo 之后重新调度任务

Timeout Policy

RETRY :再次重试任务
TIME_OUT_WF : 工作流程被标记为TIMED_OUT并终止
ALERT_ONLY : 注册计数器（task_timeout）

Workflow Definition

工作流程使用基于JSON的DSL进行定义。

Example

{

  "name": "encode_and_deploy",

  "description": "Encodes a file and deploys to CDN",

  "version": 1,

  "tasks": [

    {

      "name": "encode",

      "taskReferenceName": "encode",

      "type": "SIMPLE",

      "inputParameters": {

        "fileLocation": "${workflow.input.fileLocation}"

      }

    },

    {

      "name": "deploy",

      "taskReferenceName": "d1",

      "type": "SIMPLE",

      "inputParameters": {

        "fileLocation": "${encode.output.encodeLocation}"

      }

    }

  ],

  "outputParameters": {

    "cdn_url": "${d1.output.location}"

  },

  "schemaVersion": 2

}

name：工作流的名字
description： 工作流的描述性名称
version：用于标识模式版本的数字字段。 使用增长的数字，执行工作流时如果没有指定，就用最高版本
tasks： 一系列任务定义，如下所述。
outputParameters：用于生成工作流输出的JSON模板，如果未指定，则将输出定义为最后一个执行的任务的输出
inputParameters：输入参数列表 用于记录工作流所需的输入，可选的

Tasks within Workflow(工作流内的任务)

工作流中的tasks属性定义了要按该顺序执行的任务数组。以下是每个任务所需的强制性最低参数：

name：任务名称在启动工作流之前，必须将其注册为Conductor的任务类型

taskReferenceName：别名用于引用工作流中的任务。必须是唯一的。

type：任务类型 SIMPLE用于由远程工作人员执行的任务或系统任务类型之一

optional： true或false(默认为false)。当设置为true时 - 即使任务失败，工作流仍然继续。该任务的状态反映为COMPLETED_WITH_ERRORS

inputParameters: JSON模板定义了给定任务的输入

除了这些参数之外，还需要特定于任务类型的附加参数，如文中所述

Wiring Inputs and Outputs

当触发新的执行时，工作流由客户端输入。工作流输入是一个JSON有效载荷，可通过$ {workflow.input ...}表达式获得。

基于在工作流定义中配置的inputParameters模板，给出工作流中的每个任务的输入。 inputParameters是一个JSON片段，其值包含用于在执行期间从工作流或另一个任务的输入或输出值映射值的参数。
映射值的语法遵循以下模式：
${SOURCE.input/output.JSONPath}
SOURCE：可以是“工作流程”或任何任务的reference name
input/output：指源的输入或输出
JSONPath： JSON路径表达式从源输入/输出中提取JSON片段     http://goessner.net/articles/JsonPath/
conductor支持JSONPath规范，并从这里使用Java实现。
https://github.com/json-path/JsonPath

例如：
考虑一个任务，使用工作流中的输入/输出参数 和 名为loc_task的任务为输入配置。

{

  "inputParameters": {

    "movieId": "${workflow.input.movieId}",

    "url": "${workflow.input.fileLocation}",

    "lang": "${loc_task.output.languages[0]}",

    "http_request": {

      "method": "POST",

      "url": "http://example.com/${loc_task.output.fileId}/encode",

      "body": {

        "recipe": "${workflow.input.recipe}",

        "params": {

          "width": 100,

          "height": 100

        }

      },

      "headers": {

        "Accept": "application/json",

        "Content-Type": "application/json"

      }

    }

  }

}

将以下内容作为工作流输入

{

  "movieId": "movie_123",

  "fileLocation":"s3://moviebucket/file123",

  "recipe":"png"

}

而loc_task的输出如下所示;

{

  "fileId": "file_xxx_yyy_zzz",

  "languages": ["en","ja","es"]

}

在调度任务时，Conductor将合并来自工作流输入和loc_task输出的值，并将其输入到任务中，如下所示：

{

  "movieId": "movie_123",

  "url": "s3://moviebucket/file123",

  "lang": "en",

  "http_request": {

    "method": "POST",

    "url": "http://example.com/file_xxx_yyy_zzz/encode",

    "body": {

      "recipe": "png",

      "params": {

        "width": 100,

        "height": 100

      }

    },

    "headers": {

        "Accept": "application/json",

        "Content-Type": "application/json"

    }

  }

}

巴特西

conductor元数据定义

Task Definition

Workflow Definition

Tasks within Workflow(工作流内的任务)

Wiring Inputs and Outputs

最新文章

热门文章