在 Apache DolphinScheduler& Apache ShenYu(Incubating) Meetup 上，YY 直播 软件工程师 袁丙泽 为我们分享了《YY 直播基于 Apache DolphinScheduler 的适配与探索》。

本次演讲主要包括四个部分：

• YY 直播引入 Apache DolphinScheduler 的背景

• Apache DolphinScheduler 的引入过程

• Apache DolphinScheduler 应用的适配

• YY 直播未来的规划

讲师介绍

袁丙泽

YY 直播 软件工程师，10 余年工作经验，主要从事风控大数据平台开发工作，对常用大数据组件深感兴趣，研发经验丰富

背景

YY 直播是中国领先的语音社交视频直播企业，目前我们团队的主要职责是保障公司的业务安全。

01 技术现状

目前我们采用分层的技术架构，最底层是数据源层，其次从下往上依次是采集层、存储层和管理层和计算层与应用层。 

在数据源层，我们目前会去拉取各个业务方的一个关系型数据库数据，以及通过 API 向我们传输的数据，还有一些数据是通过 Kafka 这种流的方式来传输给我们。

采集层采用了我们自己研发的一套数据采集系统。

存储层中，我们目前将数据主要放在了关系型数据库中，如 Clickhouse，还有一小部分会放在一些非关系型数据库中，如 Redis 和图库。当然大部分数据都存储在大数据系统中。

管理层我们主要有大数据管理系统，结合自己研发的一个计算调度以及任务管理系统和服务治理平台。

02 调度 Apache DolphinScheduler 之前的问题

1、调度平台复杂：团队除了有基于 Xxl-job 的任务调度外，部分老项目中有使用 Crontab、Springboot、Scheduler、Quartz 等管理任务的启动。

2、任务依赖需求强烈：目前我们所使用的的调度，仅能设置单个任务的执行，无法通过任务依赖形成工作流，任务依赖设置严重依赖于个人经验设定定时时间。实际上很多任务都需要有依赖关系。

3、任务复杂多样：目前任务有基于大数据系统的 Spark、Flink 任务，服务治理平台中各种 Java 服务任务、Shell、Java application、Python 等。

引入过程

在需求调研中，我们实际上需要一款调度平台，需要满足如下条件：

1、统一管理任务及依赖关系

随着业务计算的需求越来越多，特别是各种各样的画像计算和任务，这些任务分散在各个系统当中，管理起来非常困难，部分任务之间有一定的依赖关系，但配置其时间依靠的是个人经验。急需一款能够统一配置管理依赖的产品。

2、兼容公司内部各平台系统

我们需要调度任务平台管理我们的任务，同时为了快速投入使用，调度平台需要兼容我们公司其他的平台系统，如内部的 Datax 和 Crontab 服务。

3、高可用、高性能、高并发，容易使用

最后为了保证业务的稳定性，我们也需要这种调度平台能够高可用、高性能、高并发，并且容易使用。

通过调研我们发现，Apache DolphinScheduler 几乎就是为我们设计的，适配过程中无需太多修改，就能满足我们需求。

应用适配

Apache DolphinScheduler 是一个分布式去中心化，易扩展的可视化 DAG 工作流任务调度系统，致力于解决数据处理流程中错综复杂的依赖关系，使调度系统在数据处理流程中开箱即用，这非常符合我们的需求。

首先了解下 Apache DolphinScheduler 的架构，便于理解接下来的适配案例。

Apache DolphinScheduler 主要有 API、master、 worker、 log 以及 alert 这 5 个模块。

API 接口层，主要负责处理前端 UI 层的请求。该服务统一提供 RESTful api 向外部提供请求服务。接口包括工作流的创建、定义、查询、修改、发布、下线、手工启动、停止、暂停、恢复、从该节点开始执行等等。

MasterServer 采用分布式无中心设计理念，MasterServer 主要负责 DAG 任务切分、任务提交监控，并同时监听其它 MasterServer 和 WorkerServer 的健康状态。MasterServer 服务启动时向 Zookeeper 注册临时节点，通过监听 Zookeeper 临时节点变化来进行容错处理。

WorkerServer 也采用分布式无中心设计理念，WorkerServer 主要负责任务的执行和提供日志服务。WorkerServer 服务启动时向 Zookeeper 注册临时节点，并维持心跳。workServer 还提供有 logger 服务。

Alert 提供告警相关接口，接口主要包括两种类型的告警数据的存储、查询和通知功能。其中通知功能又有邮件通知和**SNMP(暂未实现)**两种。

目前我们部署的是 2.0 版本，主要使用了 4 台物理机，在这 4 台物理机上部署了 2 个 master 实例，2 个 API 实例和 3 个 worker 与 logger 实例，一个 alert 实例。

接下来分享 3 个具体的适配案例。

首先是与我们服务治理平台的适配，该适配主要目的是用于任务监控；尽管 Apache DolphinScheduler 本身提供有任务监控模块，我们同事早已经习惯利用服务治理平台统一管理监控。所以我们需要把 Apache DolphinScheduler 任务运行状态及时上报至服务治理平台。

01 服务治理适配—MasterServer 服务说明

在适配之前，再次详细了解下 MasterServer 服务，MasterServer 提供有：

1. Distributed Quartz 分布式调度组件，主要负责定时任务的启停操作，当 quartz 挑起任务后，Master 内部会有线程池具体负责处理任务的后续操作；

2. MasterSchedulerThread 是一个扫描线程，定时扫描数据库中的 command 表，根据不同的命令类型进行不同的业务操作；

3. MasterExecThread（WorkflowExecutThread.java）主要负责 DAG 任务切分、任务提交监控、各种不同命令类型的逻辑处理；

4. MasterTaskExecThread 主要负责任务的持久化。

02 服务治理适配-code

我们的需求是监控任务，通过代码分析，我们发现任务提交与监听主要在 WorkflowExecuteThread 类中的方法中实现，该类会启动多个实例线程。分别负责任务执行与监听。其流程图如下：

任务提交及监控流程图

我们的需求是监控任务，通过分析代码后发现，WorkflowExecuteThread 主要有 startprocess 和 handle events 两个方法分别实现了任务执行与监听。其实我们主要在 handleEvents 方法中注入我们的服务治理平台数据收集代码，这样就能把任务监听情况及时上报到我们服务治理平台了。

其修改部分如下：

在服务治理平台中具体的效果图如下：

除了监控我们的具体任务状况外，我们还会分 project 去做一些监控，最后都通过服务治理平台来做监控操作，比如像一些任务如果比较重要，我们就会配置一些电话报警，即一旦这个任务失败或者未按时执行完毕，便会进行电话通知。

03 Datax 服务适配过程

第 2 个案例是关于 Datax 服务的适配过程。我们在研究 Apache DolphinScheduler 的时候，发现其已经集成了 Datax 类型的任务，这个对我们非常友好。因为我们也有数量相当多的任务是通过 Datax 来实现的，并且我们也开发了一部分 Datax 的插件，来去适配内部各个系统与存储的数据读写。

Datax 适配的时候主要分为两部分，一部分是通过这种自定义模板来去实现，这部分其实就是我们将之前的一些 Datax 的服务拷贝过来，稍加修改，就能够实现了，主要涉及到的是一些非关型数据库之间的一些数据交互。

而纯粹的关型数据库之间的交互，我们还是需要通过配置方式实现。

首先我们在配置 Clickhouse 读写任务时，就遇见了一个小 bug。

04 Datax 服务适配—Clickhouse 兼容 #8092

我们在使用 Datax 来读取 Clickhouse 数据源的数据时，发现在 sql 当中，只要引用参数，无论时间参数还是其他参数，在提交的时都会失败，我们就怀疑其中可能有一些 bug，阅读错误日志的时候，也发现在 Apache DolphinScheduler 提交 SQL 时，是参数并未被替换就直接提交给了 Clickhouse 去执行，由于 clickhouse 并不能识别我们的 Apache DolphinScheduler 参数,所以就直接抛出异常了。我们梳理了一下 Apache DolphinScheduler 在执行 datax 任务时读取 clickhouse 的流程。其中在将我们在 Apache DolphinScheduler 配置转为 datax 配置流程如下：

系统首先要做的就是先去解析 sql 的所有语法，然后通过语法拿到一些列的信息，这时它要去调用 sql 解析器。在这个过程当中，如果 Apache DolphinScheduler 没有对我们的这个参数去做替换，在执行这个 circle 的时候就会发生错误，最后导致整个任务失败。

因此在解决的过程中，既然可能获取不到 Clickhouse 的解析器，最好的方法就是直接加入一个解析器。首先构建一个 Json 文件，然后格式化解析出来的所有的链，最后对语法去做一次解析，层层调用，最后能够调用到目标解析器。

05 Time 参数适配 Apache DolphinScheduler 现状

最后的案例是关于时间参数适配。

Apache DolphinScheduler 虽然提供有时间参数，但是我们自己的数据大部分都需要精确到毫秒级别的 unixtime 时间。通过阅读 Apache DolphinScheduler 的文档，我们遗憾地发现其并未提供该类型时间参数的实现。翻阅源码过程中，我们发现 Apache DolphinScheduler 提供有 timestamp 函数，实际上能够提供 unixtime 时间值。

在使用 timestamp 的时候，我们发现有两个小问题，首先 timestamp 直接表达 unixtime 有一些歧义，其次 timestamp 仅支持到秒级别，而我们大部分数据需要毫秒级别。为了方便使用，我们对此部分做了一些修改进行适配。

适配过程

首先我们做的第一件事情就是消除歧义，在 Apache DolphinScheduler 中，Timestamp 是表达时间的方式，从 Wiki 百科获得的关于 Timestamp 和 Unix time 时间表达的解释能看出，Timestamp 通常是通过日期加时间来表示的，但是 Unix time 时间采用的是格林威治时间，从 1970 年 1 月 1 日零时零分零秒至今，并且不考虑微秒的时间表达，采用的是整数。

明确了需求，接下来就需要了解如何实现了。我们通过分析代码发现，时间参数函数的实现是通过 api 方式层层调用，最终主要函数均通过在 TimePlaceHolderUtils 类中 calculateTime 的方法实现。该方法实现过程中，也会调用 TaskConstants 类中的表达时间函数名称的常量。于是我们对其中 TaskConstants 类的一些常量进行了修改。又因为我们需要毫秒级别的函数，加入了一个 milli_unixtime 函数，最后为了满足设备用户的需求，我们加入了一些更精度更高的函数，如微秒和纳秒的函数。

 在补数功能上，在使用 Apache DolphinScheduler 之后，我们只需要在手动执行任务的时候选中补数的功能，再填充上我们要调度的日期，就可以直接进行补充了，同时我们还可以填写并行度。这个功能对我们这来说非常实用的，在 Apache DolphinScheduler 2.0 版本以后，时间的配置和执行的时间有日绩差的问题也被解决，在使用上带来了很大的便利。

未来规划

在使用的过程中，我们发现通过 Apache DolphinScheduler 配置的任务在使用数据源方面，目前还不支持高可用的方案，这个需求在我们这里是比较强烈的，因此目前我们也正在做高可用的适配。

其次，我们目前使用的是 Apache DolphinScheduler 的 2.0 版本，因为社区比较活跃，版本升级也比较快，即使是一个小版本的升级，也会带来一些很大的功能和设计上的一些变化。比如在新版本当中，告警功能已经插件化，也解决了一些补数日期换算的问题。这也驱动着我们团队升级到新的版本去体验一些新的功能。虽然目前 Apache DolphinScheduler 只是在我们自己的小团队内部使用，但我们也正在思考让整个公司普遍使用的可行性方案。

尽管 Apache DolphinScheduler 非常完美地解决我们的大部分问题，并且大幅度提高我们的工作效率。但在各种复杂的情况下，我们还是会遇见一些小的 Bug，我们未来也会在修复后提交给官方，当然我们自己在使用过程中也尝试了一些小 Future，未来也会提交给官方共同讨论。

参与贡献

随着国内开源的迅猛崛起，Apache DolphinScheduler 社区迎来蓬勃发展，为了做更好用、易用的调度，真诚欢迎热爱开源的伙伴加入到开源社区中来，为中国开源崛起献上一份自己的力量，让本土开源走向全球。

贡献第一个 PR(文档、代码) 我们也希望是简单的，第一个 PR 用于熟悉提交的流程和社区协作以及感受社区的友好度。

社区汇总了以下适合新手的问题列表：https://github.com/apache/dolphinscheduler/issues/5689

非新手问题列表：https://github.com/apache/dolphinscheduler/issues?q=is%3Aopen+is%3Aissue+label%3A%22volunteer+wanted%22

如何参与贡献链接：https://dolphinscheduler.apache.org/zh-cn/docs/development/contribute.html

来吧，DolphinScheduler 开源社区需要您的参与，为中国开源崛起添砖加瓦吧，哪怕只是小小的一块瓦，汇聚起来的力量也是巨大的。

参与开源可以近距离与各路高手切磋，迅速提升自己的技能，如果您想参与贡献，我们有个贡献者种子孵化群，可以添加社区小助手微信(Leonard-ds) ，手把手教会您( 贡献者不分水平高低，有问必答，关键是有一颗愿意贡献的心 )。