本文介绍 Checker 类型的 Activity 配置。当前 job-manager 中,Checker 用于在 Pipeline 中轮询等待上游 Pipeline 或上游 Activity 达到完成状态,满足条件后放行下游 Activity 继续执行。
Checker Activity(YAML 类型标识为 checker)通过轮询检测依赖是否满足,是实现跨 Pipeline 依赖和跨项目协同调度的控制节点类型。
适用场景:
projectId 指定上游 Pipeline 所在项目;未配置时默认使用当前 Job 外部上下文中的 projectId。当前不支持的内容:
checkerConfig.checks 结构。projectName 字段,应使用 projectId。type: sensor、sensorType: las_catalog、sensorType: tos 等 Sensor 配置。- name: wait_for_order_etl type: checker checkIntervalInSeconds: 60 maxCheckTimes: 120 dependsOn: pipelines: - name: daily_order_etl projectId: 12345 offsetsType: NEAREST offsets: - 0 position: x: "200" y: "100"
说明
dependsOn.pipelines[].name 表示上游 Pipeline / 调度任务 BizId。projectId 表示上游 Pipeline 所在项目 ID。offsetsType 当前建议固定为 NEAREST。offsets 使用整数数组,例如 [0] 表示同周期上游实例。- name: wait_for_transform type: checker checkIntervalInSeconds: 60 maxCheckTimes: 120 dependsOn: externalActivities: - pipelineName: daily_order_etl name: transform_orders projectId: 12345 offsetsType: NEAREST offsets: - 0 position: x: "200" y: "100"
说明
dependsOn.externalActivities[].pipelineName 表示上游 Pipeline 名称 / 调度任务 BizId。dependsOn.externalActivities[].name 表示上游 Pipeline 内的 Activity 名称。activityName 字段;原文中的 activityName 应改为 name。一个 Checker 可同时检查多个上游。所有依赖均满足后,Checker 才会成功并放行下游。
- name: wait_for_multi_upstreams type: checker checkIntervalInSeconds: 120 maxCheckTimes: 90 dependsOn: pipelines: - name: order_etl projectId: 123 offsetsType: NEAREST offsets: [0] - name: user_etl projectId: 123 offsetsType: NEAREST offsets: [0] externalActivities: - pipelineName: traffic_etl name: aggregate_traffic projectId: 456 offsetsType: NEAREST offsets: [0]
字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| String | 是 | 固定为 |
| Object | 是 | Checker 的依赖配置。Checker 至少需要配置 |
| Integer | 否 | 轮询检查间隔,默认 60 秒。 |
| Integer | 否 | 最大检查次数,默认 3000。达到上限后 Checker 置为 |
| Object | 否 | 节点在画布中的位置。 |
dependsOn.pipelines 字段 用于等待整个上游 Pipeline 完成。
字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| String | 是 | 上游 Pipeline 名称 / 调度任务 BizId。 |
| Long | 否 | 上游 Pipeline 所在项目 ID;不填时使用当前 Job 外部上下文中的 |
| String | 否 | 当前建议使用 |
| Array | 否 | 偏移列表,例如 |
dependsOn.externalActivities 字段 用于等待上游 Pipeline 中的某个 Activity 完成。
字段 | 类型 | 必填 | 说明 |
|---|---|---|---|
| String | 是 | 上游 Pipeline 名称 / 调度任务 BizId。 |
| String | 是 | 上游 Pipeline 内的 Activity 名称。 |
| Long | 否 | 上游 Pipeline 所在项目 ID;不填时使用当前 Job 外部上下文中的 |
| String | 否 | 当前建议使用 |
| Array | 否 | 偏移列表,例如 |
当前 Checker 与校验规则建议使用:
offsetsType: NEAREST offsets: - 0
Checker 的轮询行为由以下两个字段控制:
checkIntervalInSeconds: 60 maxCheckTimes: 120
含义:
checkIntervalInSeconds 秒检查一次依赖状态。maxCheckTimes 次。maxCheckTimes 后依赖仍未满足,Checker 会置为 FAILED。如果希望表达“最多等待 7200 秒”,可配置为:
checkIntervalInSeconds: 60 maxCheckTimes: 120
Checker 判断调度侧上游 Pipeline 状态时,依赖调度返回的 StatusAlias 字段。
成功状态包括:
succeededforce_succeedsucceeded_downstreamsucceed_dual_runsucceed_dual_run_downstream失败状态包括:
failedfailed_retrycanceledaborteddeleted运行中或等待中的状态会继续轮询。
Checker 节点开始 → 按 checkIntervalInSeconds 间隔轮询 → 检查 dependsOn.activities / dependsOn.pipelines / dependsOn.externalActivities → 全部满足 → SUCCEEDED → 下游继续执行 → 部分未满足 → RUNNING → 继续等待 → 上游失败 / 取消 / 删除 → FAILED → 达到 maxCheckTimes → FAILED
建议 | 说明 |
|---|---|
使用 | 当前实现不读取 |
使用 | 当前模型和运行时只识别 |
检查整个 Pipeline 用 | 字段名为 |
检查外部 Activity 用 |
|
控制等待时间用 | 当前 Checker 未实现 |
暂不写 Sensor 配置 | 当前代码没有 |