Apache DolphinScheduler新一代分布式工作流任务调度平台实战-中

• MasterServer：MasterServer采用分布式无中心设计理念，MasterServer主要负责 DAG 任务切分、任务提交监控，并同时监听其它MasterServer和WorkerServer的健康状态。 MasterServer服务启动时向Zookeeper注册临时节点，通过监听Zookeeper临时节点变化来进行容错处理。 MasterServer基于netty提供监听服务。该服务主要包含：
  • Distributed Quartz分布式调度组件，主要负责定时任务的启停操作，当quartz调起任务后，Master内部会有线程池具体负责处理任务的后续操作
  • MasterSchedulerThread是一个扫描线程，定时扫描数据库中的 command 表，根据不同的命令类型进行不同的业务操作
  • MasterExecThread主要是负责DAG任务切分、任务提交监控、各种不同命令类型的逻辑处理
  • MasterTaskExecThread主要负责任务的持久化
• WorkerServer：WorkerServer也采用分布式无中心设计理念，WorkerServer主要负责任务的执行和提供日志服务。 WorkerServer服务启动时向Zookeeper注册临时节点，并维持心跳。 Server基于netty提供监听服务。Worker该服务主要包含：
  • FetchTaskThread主要负责不断从Task Queue中领取任务，并根据不同任务类型调用TaskScheduleThread对应执行器。
• ZooKeeper：ZooKeeper服务，系统中的MasterServer和WorkerServer节点都通过ZooKeeper来进行集群管理和容错。另外系统还基于ZooKeeper进行事件监听和分布式锁。 我们也曾经基于Redis实现过队列，不过我们希望DolphinScheduler依赖到的组件尽量地少，所以最后还是去掉了Redis实现。
• Task Queue：提供任务队列的操作，目前队列也是基于Zookeeper来实现。由于队列中存的信息较少，不必担心队列里数据过多的情况，实际上我们压测过百万级数据存队列，对系统稳定性和性能没影响。
• Alert：提供告警相关接口，接口主要包括告警两种类型的告警数据的存储、查询和通知功能。其中通知功能又有邮件通知和SNMP(暂未实现)两种。
• API：API接口层，主要负责处理前端UI层的请求。该服务统一提供RESTful api向外部提供请求服务。 接口包括工作流的创建、定义、查询、修改、发布、下线、手工启动、停止、暂停、恢复、从该节点开始执行等等。
• UI：系统的前端页面，提供系统的各种可视化操作界面。

启动流程图

架构设计思想简述

DolphinScheduler架构和实现中有非常多的优秀设计思想，详细可以查阅官方说明，下面是核心点简述：

• 去中心化vs中心化
  • DolphinScheduler的去中心化是Master/Worker注册到Zookeeper中，实现Master集群和Worker集群无中心，并使用Zookeeper分布式锁来选举其中的一台Master或Worker为“管理者”来执行任务。
• 分布式锁实践
  • DolphinScheduler使用ZooKeeper分布式锁来实现同一时刻只有一台Master执行Scheduler，或者只有一台Worker执行任务的提交。
• 线程不足循环等待问题
  • 如果一个DAG中没有子流程，则如果Command中的数据条数大于线程池设置的阈值，则直接流程等待或失败。
  • 如果一个大的DAG中嵌套了很多子流程，则会产生“死等”状态。增加一种资源不足的Command类型，如果线程池不足，则将主流程挂起。这样线程池就有了新的线程，可以让资源不足挂起的流程重新唤醒执行。注意：Master Scheduler线程在获取Command的时候是FIFO的方式执行的。
• 容错设计
  • 容错分为服务宕机容错和任务重试，服务宕机容错又分为Master容错和Worker容错两种情况。
• 任务优先级设计
  • 按照不同流程实例优先级优先于同一个流程实例优先级优先于同一流程内任务优先级优先于同一流程内任务提交顺序依次从高到低进行任务处理。
    • 流程定义的优先级是考虑到有些流程需要先于其他流程进行处理，这个可以在流程启动或者定时启动时配置，共有5级，依次为HIGHEST、HIGH、MEDIUM、LOW、LOWEST。
    • 任务的优先级也分为5级，依次为HIGHEST、HIGH、MEDIUM、LOW、LOWEST。
• Logback和netty实现日志访问
  • DolphinScheduler的轻量级性，所以选择了gRPC实现远程访问日志信息。
  • 使用自定义Logback的FileAppender和Filter功能，实现每个任务实例生成一个日志文件。

• DolphinScheduler-Master 分配任务至 worker,默认配置为线性加权负载。由于路由是在客户端做的，即 master 服务，可以更改 master.properties 中的 master.host.selector 来配置算法。eg：master.host.selector=random（不区分大小写）；提供了三种算法:

  • 加权随机（random）
  • 平滑轮询（roundrobin）
  • 线性负载（lowerweight）

• Worker 负载均衡配置：配置文件 worker.properties，权重

  • 上述所有的负载算法都是基于权重来进行加权分配的，权重影响分流结果。你可以在 修改 worker.weight 的值来给不同的机器设置不同的权重。
  • 预热：考虑到 JIT 优化，我们会让 worker 在启动后低功率的运行一段时间，使其逐渐达到最佳状态，这段过程我们称之为预热。因此 worker 在启动后，他的权重会随着时间逐渐达到最大（默认十分钟，我们没有提供配置项，如果需要，你可以修改并提交相关的 PR）

• 负载均衡算法细述

  • 随机（加权）：该算法比较简单，即在符合的 worker 中随机选取一台（权重会影响他的比重）。
  • 平滑轮询（加权）：加权轮询算法一个明显的缺陷。即在某些特殊的权重下，加权轮询调度会生成不均匀的实例序列，这种不平滑的负载可能会使某些实例出现瞬时高负载的现象，导致系统存在宕机的风险。为了解决这个调度缺陷，我们提供了平滑加权轮询算法。每台 worker 都有两个权重，即 weight（预热完成后保持不变），current_weight（动态变化），每次路由。都会遍历所有的 worker，使其 current_weight+weight，同时累加所有 worker 的 weight，计为 total_weight，然后挑选 current_weight 最大的作为本次执行任务的 worker，与此同时，将这台 worker 的 current_weight-total_weight。
  • 线性加权(默认算法)：该算法每隔一段时间会向注册中心上报自己的负载信息。
    • 我们主要根据两个信息来进行判断
      • load 平均值（默认是 CPU 核数 *2）
      • 可用物理内存（默认是 0.3，单位是 G）
    • 如果两者任何一个低于配置项，那么这台 worker 将不参与负载。（即不分配流量），可以在 worker.properties 修改下面的属性来自定义配置
      • worker.max.cpuload.avg=-1 (worker最大cpuload均值，只有高于系统cpuload均值时，worker服务才能被派发任务. 默认值为-1: cpu cores * 2)
      • worker.reserved.memory=0.3 (worker预留内存，只有低于系统可用内存时，worker服务才能被派发任务，单位为G)

  由于在master-server调度过程中，会产生大量的数据库读取操作，如tenant，user，processDefinition等，一方面对DB产生很大的读压力，另一方面则会使整个核心调度流程变得缓慢；考虑到这部分业务数据是读多写少的场景，故引入了缓存模块，以减少DB读压力，加快核心调度流程；缓存模块采用spring-cache机制，可直接在spring配置文件中配置是否开启缓存（默认none关闭）, 缓存类型；

  • 目前采用caffeine进行缓存管理，可自由设置缓存相关配置，如缓存大小、过期时间等；
  • 缓存读取：缓存采用spring-cache的注解，配置在相关的mapper层，可参考如：TenantMapper.
  • 缓存更新：业务数据的更新来自于api-server, 而缓存端在master-server, 故需要对api-server的数据更新做监听(aspect切面拦截@CacheEvict)，当需要进行缓存驱逐时会通知master-server，master-server接收到cacheEvictCommand后进行缓存驱逐；
  • 需要注意的是：缓存更新的兜底策略来自于用户在caffeine中的过期策略配置，请结合业务进行配置；

参数优先级

DolphinScheduler 中所涉及的参数值的定义可能来自三种类型：

• 全局参数：在工作流保存页面定义时定义的变量
• 上游任务传递的参数：上游任务传递过来的参数
• 本地参数：节点的自有变量，用户在“自定义参数”定义的变量，并且用户可以在工作流定义时定义该部分变量的值

因为参数的值存在多个来源，当参数名相同时，就需要会存在参数优先级的问题。DolphinScheduler 参数的优先级从高到低为：本地参数 > 上游任务传递的参数 > 全局参数

在上游任务传递的参数中，由于上游可能存在多个任务向下游传递参数，当上游传递的参数名称相同时：

• 下游节点会优先使用值为非空的参数
• 如果存在多个值为非空的参数，则按照上游任务的完成时间排序，选择完成时间最早的上游任务对应的参数

基础内置参数

衍生内置参数

• 支持代码中自定义变量名，声明方式：${变量名}。可以是引用 "系统参数"

• 我们定义这种基准变量为 [...]格式的，[...]格式的，[yyyyMMddHHmmss] 是可以任意分解组合的，比如：[yyyyMMdd],[yyyyMMdd],[HHmmss], $[yyyy-MM-dd] 等

• 也可以通过以下两种方式：

  • 使用add_months()函数，该函数用于加减月份， 第一个入口参数为[yyyyMMdd]，表示返回时间的格式 第二个入口参数为月份偏移量，表示加减多少个月

    • 后 N 年：$[add_months(yyyyMMdd,12*N)]
    • 前 N 年：$[add_months(yyyyMMdd,-12*N)]
    • 后 N 月：$[add_months(yyyyMMdd,N)]
    • 前 N 月：$[add_months(yyyyMMdd,-N)]

  • 直接加减数字 在自定义格式后直接“+/-”数字

    • 后 N 周：$[yyyyMMdd+7*N]
    • 前 N 周：$[yyyyMMdd-7*N]
    • 后 N 天：$[yyyyMMdd+N]
    • 前 N 天：$[yyyyMMdd-N]
    • 后 N 小时：$[HHmmss+N/24]
    • 前 N 小时：$[HHmmss-N/24]
    • 后 N 分钟：$[HHmmss+N/24/60]
    • 前 N 分钟：$[HHmmss-N/24/60]

本地参数和全局参数

本地参数的作用域：在任务定义页面配置的参数，默认作用域仅限该任务，如果配置了参数传递则可将该参数作用到下游任务中。

使用前面shell演示工作流定义，在shell-nodeA添加本地参数:

• dt：参数名
• IN：IN 表示局部参数仅能在当前节点使用，OUT 表示局部参数可以向下游传递
• DATE：数据类型，日期
• $[yyyy-MM-dd]：自定义格式的衍生内置参数

全局参数作用域：全局参数是指针对整个工作流的所有任务节点都有效的参数，在工作流定义页面配置。本地任务引用全局参数的前提是已经定义了全局参数，使用方式和本地参数中的使用方式类似，但是参数的值需要配置成全局参数中的 key。

在shell-nodeB和shell-nodeC中的脚本输出echo ${dt},然后点击保存工作流，添加全局变量

从任务实例点击右边查看日志，可以看到shell-nodeA输出的是当天日期2022-08-02，而shell-nodeB和shell-nodeC输出的是前一天日期2022-08-01。

工作流传参

DolphinScheduler 允许在任务间进行参数传递，目前传递方向仅支持上游单向传递给下游。目前支持这个特性的任务类型有Shell、SQL、Procedure。

当定义上游节点时，如果有需要将该节点的结果传递给有依赖关系的下游节点，需要在【当前节点设置】的【自定义参数】设置一个方向是 OUT 的变量。目前我们主要针对 SQL 和 SHELL 节点做了可以向下传递参数的功能。

• 注：若节点之间没有依赖关系，则局部参数无法通过上游传递。

下面通过 SHELL 任务来创建本地参数并赋值传递给下游，用户需要传递参数，在定义 SHELL 脚本时，需要输出格式为 ${setValue(key=value)} 的语句，key 为对应参数的 prop，value 为该参数的值。在shell-nodeA任务节点中的自定义参数中添加设置参数传递如下:

在shell-nodeC中脚本输出echo ${transfer}，保存工作流定义-上线-运行，查看shell-nodeC任务示例的日志，可以得到输出了20220701达到工作流参数。

数据源管理

支持数据源

数据源中心支持MySQL、POSTGRESQL、HIVE/IMPALA、SPARK、CLICKHOUSE、ORACLE、SQLSERVER等数据源。

• 点击"数据源中心->创建数据源"，根据需求创建不同类型的数据源
• 点击"测试连接"，测试数据源是否可以连接成功（只有当数据源通过连接性测试后才能保存数据源）。

以 MySQL 为例，如果想要使用 MySQL 数据源，需要先在 mysql maven 仓库 中下载对应版本的 JDBC 驱动，将其移入 api-server/libs 以及 worker-server/libs 文件夹中，最后重启 api-server 和 worker-server 服务，即可使用 MySQL 数据源。如果你使用容器启动 DolphinScheduler，同样也是将 JDBC 驱动挂载放到以上两个服务的对应路径下后，重启驱动即可。

创建MySQL数据源

由于前面部署DolphinScheduler集群已将MySQL的驱动复制到所有节点master和worker、api-server、alert-server上了，因此这里可以开始创建MySQL数据源

• 数据源：选择 MYSQL
• 数据源名称：输入数据源的名称
• 描述：输入数据源的描述
• IP 主机名：输入连接 MySQL 的 IP
• 端口：输入连接 MySQL 的端口
• 用户名：设置连接 MySQL 的用户名
• 密码：设置连接 MySQL 的密码
• 数据库名：输入连接 MySQL 的数据库名称
• Jdbc 连接参数：用于 MySQL 连接的参数设置，以 JSON 形式填写

jdbc连接参数如下：

{
    "useSSL": "false",
    "useUnicode": "true",
    "characterEncoding": "utf-8",
    "allowMultiQueries": "true",
    "zeroDateTimeBehavior": "convertToNull",
    "allowPublicKeyRetrieval": "true"
}

点击数据源记录的数据源参数

创建ClickHouse数据源

这里我们使用前面部署好的ClickHouse，CLICKHOUSE数据源驱动原生已支持

工作流实践

SQL工作流

拖拉SQL引擎图标，创建名称为sql_node1任务定义

insert into table01 
values('1001',99.9,'2022-08-02 22:00:00');

CREATE TABLE table01
(
    `id` String,
    `price` Float64,
    `create_time` DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(create_time)
ORDER BY id;

创建名称为sql_node2任务定义

insert into table02 
values('1002',199.9,'2022-08-02 23:00:00');

CREATE TABLE table02
(
    `id` String,
    `price` Float64,
    `create_time` DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(create_time)
ORDER BY id;

创建名称为sql_node3任务定义

insert into table03 select * from table01 
union all select * from table02;

CREATE TABLE table03
(
    `id` String,
    `price` Float64,
    `create_time` DateTime
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(create_time)
ORDER BY id;

编辑好工作流的依赖关系，sql_node1和sql_node2都完成后再执行sql_node3，保存名称为sql演示工作流。

上线sql演示工作流并点击执行，查询工作流实例执行结果

查询任务实例执行结果

使用ClickHouse的客户端登录ClickHouse查询数据，已经显示正确的结果

工作流定时

大部分任务都会有定时运行的需求，这就需要定时工作流，创建步骤：点击项目管理->工作流->工作流定义，进入工作流定义页面，上线工作流，点击"定时"按钮,弹出定时参数设置弹框：

• 选择起止时间。在起止时间范围内，定时运行工作流；不在起止时间范围内，不再产生定时工作流实例。

• 添加一个每隔 5 分钟执行一次的定时。

• 失败策略、通知策略、流程优先级、Worker 分组、通知组、收件人、抄送人同工作流运行参数。

  点击"创建"按钮，创建定时成功，此时定时状态为"下线"，定时需上线才生效。

  定时上线：点击"定时管理"按钮，进入定时管理页面，点击"上线"按钮，定时状态变为"上线"，如下图所示，工作流定时生效。

告警模块支持场景

用户需要创建告警实例，在创建告警实例时，需要选择告警策略，有三个选项，成功发、失败发，以及成功和失败都发。在执行完工作流或任务时，如果触发告警，调用告警实例发送方法会进行逻辑判断，将告警实例与任务状态进行匹配，匹配则执行该告警实例发送逻辑，不匹配则过滤。创建完告警实例后，需要同告警组进行关联，一个告警组可以使用多个告警实例。 告警模块支持场景如下：

邮件告警示例

先准备一个邮箱，开启POP3/SMTP服务，由于是要用来发送主要是SMTP，得到授权码，保存修改。

使用管理员用户登录，进入到安全中心，选择告警实例管理，创建一个告警实例，然后选择对应的告警插件EMAIL，填写相关如下参数。

然后选择告警组管理，创建告警组，选择相应的告警实例即可。

工作流定义运行中配置通知策略为成功或失败都发，告警组为前面创建测试告警组

查看邮箱确认已收到邮件信息

**本人博客网站 **IT小神 www.itxiaoshen.com