文档中心

大数据研发治理套件

Activity （任务节点）类型

Checker（数据感知）

本文介绍 Checker 类型的 Activity 配置。当前 job-manager 中，Checker 用于在 Pipeline 中轮询等待上游 Pipeline 或上游 Activity 达到完成状态，满足条件后放行下游 Activity 继续执行。

1 概述

Checker Activity（YAML 类型标识为 checker）通过轮询检测依赖是否满足，是实现跨 Pipeline 依赖和跨项目协同调度的控制节点类型。
适用场景：

等待上游 Pipeline 完成后再执行当前 Pipeline 的下游节点。
等待上游 Pipeline 中某个特定 Activity 完成后再继续执行。
在同一个 Checker 中混合等待多个上游 Pipeline / Activity。
跨项目依赖：通过 projectId 指定上游 Pipeline 所在项目；未配置时默认使用当前 Job 外部上下文中的 projectId。

当前不支持的内容：

不支持 checkerConfig.checks 结构。
不支持 projectName 字段，应使用 projectId。
不支持 type: sensor、sensorType: las_catalog、sensorType: tos 等 Sensor 配置。
当前 Checker 只基于调度侧上游实例状态 / JM 内 Activity Job 状态做判断，不包含 LAS 分区或 TOS 文件存在性检查。

2 配置示例

2.1 检查上游 Pipeline 完成

- 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] 表示同周期上游实例。

2.2 检查上游 Pipeline 中的特定 Activity 完成

- 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。

2.3 混合检查多个上游

一个 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]

3 字段说明

3.1 Checker Activity 字段

字段	类型	必填	说明
`type`	String	是	固定为 `checker`。
`dependsOn`	Object	是	Checker 的依赖配置。Checker 至少需要配置 `activities` / `pipelines` / `externalActivities` 中的一类依赖。
`checkIntervalInSeconds`	Integer	否	轮询检查间隔，默认 60 秒。
`maxCheckTimes`	Integer	否	最大检查次数，默认 3000。达到上限后 Checker 置为 `FAILED`。
`position`	Object	否	节点在画布中的位置。

3.2 `dependsOn.pipelines` 字段

用于等待整个上游 Pipeline 完成。

字段	类型	必填	说明
`name`	String	是	上游 Pipeline 名称 / 调度任务 BizId。
`projectId`	Long	否	上游 Pipeline 所在项目 ID；不填时使用当前 Job 外部上下文中的 `projectId`。
`offsetsType`	String	否	当前建议使用 `NEAREST`。
`offsets`	Array	否	偏移列表，例如 `[0]` 表示同周期实例。

3.3 `dependsOn.externalActivities` 字段

用于等待上游 Pipeline 中的某个 Activity 完成。

字段	类型	必填	说明
`pipelineName`	String	是	上游 Pipeline 名称 / 调度任务 BizId。
`name`	String	是	上游 Pipeline 内的 Activity 名称。
`projectId`	Long	否	上游 Pipeline 所在项目 ID；不填时使用当前 Job 外部上下文中的 `projectId`。
`offsetsType`	String	否	当前建议使用 `NEAREST`。
`offsets`	Array	否	偏移列表，例如 `[0]` 表示同周期实例。

4 偏移配置

当前 Checker 与校验规则建议使用：

offsetsType: NEAREST
offsets:
  - 0

5 轮询与超时行为

Checker 的轮询行为由以下两个字段控制：

checkIntervalInSeconds: 60
maxCheckTimes: 120

含义：

每隔 checkIntervalInSeconds 秒检查一次依赖状态。
最多检查 maxCheckTimes 次。
如果达到 maxCheckTimes 后依赖仍未满足，Checker 会置为 FAILED。

如果希望表达“最多等待 7200 秒”，可配置为：

checkIntervalInSeconds: 60
maxCheckTimes: 120

6 状态判断

Checker 判断调度侧上游 Pipeline 状态时，依赖调度返回的 StatusAlias 字段。
成功状态包括：

succeeded
force_succeed
succeeded_downstream
succeed_dual_run
succeed_dual_run_downstream

失败状态包括：

failed
failed_retry
canceled
aborted
deleted

运行中或等待中的状态会继续轮询。

7 执行流程

Checker 节点开始
  → 按 checkIntervalInSeconds 间隔轮询
    → 检查 dependsOn.activities / dependsOn.pipelines / dependsOn.externalActivities
      → 全部满足 → SUCCEEDED → 下游继续执行
      → 部分未满足 → RUNNING → 继续等待
      → 上游失败 / 取消 / 删除 → FAILED
      → 达到 maxCheckTimes → FAILED

8 使用建议

建议	说明
使用 `dependsOn` 配置依赖	当前实现不读取 `checkerConfig.checks`。
使用 `projectId` 而不是 `projectName`	当前模型和运行时只识别 `projectId`。
检查整个 Pipeline 用 `dependsOn.pipelines`	字段名为 `name`，不是 `pipelineName`。
检查外部 Activity 用 `dependsOn.externalActivities`	`pipelineName` 表示上游 Pipeline，`name` 表示上游 Activity。
控制等待时间用 `checkIntervalInSeconds × maxCheckTimes`	当前 Checker 未实现 `timeout` 字段。
暂不写 Sensor 配置	当前代码没有 `sensor` ActivityType 和对应 Processor。

最近更新时间：2026.07.01 00:44:44

这个页面对您有帮助吗？

有用

无用

大数据研发治理套件

1 概述 #

2 配置示例 #

2.1 检查上游 Pipeline 完成 #

2.2 检查上游 Pipeline 中的特定 Activity 完成 #

2.3 混合检查多个上游 #

3 字段说明 #

3.1 Checker Activity 字段 #

3.2 dependsOn.pipelines 字段 #

3.3 dependsOn.externalActivities 字段 #

4 偏移配置 #

5 轮询与超时行为 #

6 状态判断 #

7 执行流程 #

8 使用建议 #