2022 年 1 月，科学技术部高新技术司副司长梅建平在“第六届中国新金融高峰论坛”上表示，当前数据量已经大大超过了处理能力的上限，若信息技术仍然是渐进式发展，则数据处理能力的提升将远远落后于指数级增长的数据量。因此，在一段时期内，数据处理能力与效率的提升仍将是大数据发展要面对的技术难点。

随着 5G、物联网等网络信息技术的快速发展以及应用的快速增长，数据量也呈指数级增长，纵观运营商整个大数据开发的链路上，在各个环节都会出现各种严峻的问题，随着数据任务调度量级日益增大，阻碍数据正确且高效地发挥价值，对运营商数据团队提出严峻挑战。

在大数据领域，越来越多的企业拥抱开源软件，在这个背景下，我们针对数据调度工具如何正确选型？

中国移动云能力中心软件开发工程师徐海辉表示：古语云“工欲善其事必先利其器”，如果你正处于观望/不知如何下手/即将参与开源项目的小伙伴，我建议可以先从一个优秀的开源社区源代码的入手，我在中国移动目前主要负责数据服务，这次在 Apache DolphinScheduler 4 月 Meetup 上为大家带来 DolphinScheduler 源码 2.X 解析，希望你有所收获。

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

• 开篇与源码环境准备

• 服务启动流程

• 任务执行流程

• 个人思考与总结

徐海辉

中国移动云能力中心 软件开发工程师。从事大数据基础平台开发，主要负责中国移动 Hadoop 大数据平台组件 Ranger 、移动云 Lake House 产品的研发

关键词：Apache DolphinScheduler 源码 2.X 解析、源码环境准备、服务启动、任务执行

Apache DolphinScheduler 源码下载链接:

https://dolphinscheduler.apache.org/zh-cn/download/download.html

01 开篇与源码环境准备

Apache DolphinScheduler 是一个基于 java 开发的开源分布式工作流调度系统。致力于可视化操作任务及工作流之间的依赖关系，并可视化整个数据流过程；解决数据处理流程中错综复杂的依赖关系。

01 入手 DolphinScheduler 需要考虑的问题

• 开源分布式的工作原理？

• 工具可视化是体现在哪里？

• Master 和 Worker 之间如何通信？

• 怎么体现工作流之间的依赖关系？

“Show me the code”

其实在这个过程中需要考虑的问题有很多，我们不妨从官方给的架构图先简单梳理一下，然后通过实际的使用，再去研究源码，一套组合拳下来就差不多了。

02 DolphinScheduler 架构图

话不多说，开撸！

4 个由源码启动的服务是：UI、MasterServer、WorkServer、AlertServer（还有 LoggerServer 图里面没有体现）

• MasterServer & WorkServer 依赖于 Zookeeper 协调服务中心注册

• MasterServer & WorkServer 1-N 组成集群，分别是单独的服务和进程

• 执行任务支持的插件：Flink、Shell、Subflow、SQL、Procedure、Python、MR、Spark、Dependent 等等

03 预先必备知识

• Netty

一个异步的、基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端

• Zookeeper

一个分布式的，开放源码的分布式应用程序协调服务，它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。

• 分布式锁

为了解决单机部署情况下的并发控制锁策略失效这个问题，需要一种跨机器的互斥机制来控制共享资源的访问

• Quartz

定时自动执行任务

• 多线程

很多小伙伴不明白 Master 和 Worker 之间是怎么通信的，DolphinScheduler 是基于 Netty 框架来实现的。这里可以拓展一下，服务器之间的通信还有很多方式：

• HTTP

HTTP：http 其实是一种网络传输协议，基于 TCP，规定了数据传输的格式。

• REST API

REST API 通信是通过 JSON 格式的字符串进行数据传输的，而字符串是可以在网络中穿透防火墙的。也就是说，REST API 可以穿透防火墙。同时字符串也可以不用受开发语言的限制，可以同时完成后端与 WEB，后端与 APP(Android，IOS)之间的通信。

• RPC

RPC 通信又称远程过程调用，在内网中速度非常快，效率高。

如下图是准备相关环境的部署，并且我还提供了一个远程 Debug 的方式，这个不难理解，比如你想调试 API，在启动过程中，它肯定会调用脚本，你只需要在脚本里面添加 Debug 启动的一些参数，就可以进行一个远程的 Debug。

像框框里面的数字实际上是服务的端口号，如果跟本地服务/其他组件端口之间有冲突的话，对它可以进行一个修改。

02 服务启动流程

01 Master 启动流程

• MasterServer 采用分布式无中心设计理念，基于 Netty 提供监听服务。

• MasterServer 服务启动时向 Zookeeper 注册临时节点，通过监听 Zookeeper 临时节点变化来进行容错处理。

• MasterServer 主要负责 DAG 任务切分、任务提交监控，并同时监听其它 MasterServer 和 WorkerServer 的健康状态。

启动步骤

1. 启动 Netty 服务端服务

2. 注册到 Zookeeper

3. 启动 Event 处理器

4. 启动 scheduler 定时任务

5. 启动 StateWheel 处理器

↓↓↓源码一览↓↓↓

对应路径里面存放了 MasterServer.java 的类，里面有对应的 main 方法

​执行完构造函数后，会启动 run 方法及其各个组件

这里会提前构造参数

调用 start 方法启动

注册元数据信息到 Zookeeper 上面，值得一提的是这里采用的是临时路径，比如说在过程中服务断开了或者 session 过期，临时路径过一段时间会自己去 Delete 掉。

启动调度服务

启动 Quartz 框架，启动完之后相当于是一个独立的服务.

小结：

• 第一步是注册这个 Netty

• 第二步是向 Zookeeper 注册数据信息。

• 第三步是启动相关组件和调度任务

• 第四步是启动 Quartz 框架

02 Worker 启动流程

• WorkerServer 采用分布式无中心设计理念，WorkerServer 基于 Netty 提供监听服务。

  
  

• WorkerServer 主要负责任务的执行和提供日志服务。WorkerServer 服务启动时向 Zookeeper 注册临时节点，并维持心跳。

  
  

拓展：防止出现断开、关闭等突发状况，这里会选择注册临时节点做出 Delete 动作。

启动步骤

1. 启动 Netty 服务端服务

2. 注册到 Zookeeper

3. 维护 workserver 节点状态

4. 启动 TaskExecuteThread

5. 启动 RetryReportTaskStatusThread

↓↓↓源码一览↓↓↓

找到 WorkServer.java 中的 main 方法，进程代码也位于 DolphinScheduler-Server 模块下面

启动 Netty 服务，过程中也会提供一些端口，以便和其他进程交互

向 Zookeeper 注册信息，其中 WorkerregistryClient 调用了 Zookeeper 的代码

主要是处理和创建一些节点

启动组件，WorkerManagerThread 主要是管理 Master 发过来的任务

启动组件，RetryReportTaskStatus 主要是给 Master 反馈信息

03 任务执行流程

首先我们可以从 UI 界面具体分析一些工作流是如何运行的？

DolphinScheduler-api :  ExecutorController.java → startProcessInstance() → createCommand() →写入表 t_ds_command

然后通过代码的跟踪去看一下

位于 DolphinScheduler-API 模块下的 controller 里面

Start-process-instance 会调用 ExecProcessInstance 方法

通过一系列的执行，最终会解析一些参数

解析完之后会生成一条记录（Create Command）

接下来 MasterServer 会不定时扫描表里面的记录，然后拉取出来，最后构建流程的实例

定时任务轮询获取

Dolphinscheduler-server: MasterSchedulerService.java → scheduleProcess() → findOneCommand() → 创建 ProcessInstance

构建 DAG 并分发到 worker

• WorkflowExecuteThread.java → startProcess() → buildFlowDag() → submitPostNode() → submitStandByTask() → submitTaskExec()

• TaskProcessor.java → dispatchTask() → 放入 TaskPriorityQueue

• TaskPriorityQueueConsumer.java: → dispatch()

• ExecutorDispatcher.java → dispatch()

• NettyExecutorManager.java → execute() →build command → doExecute() → ettyRemotingClient.send()

Master 启动完之后会不停拉取命令，然后解析完之后分发给 Worker

可以看一下它的 run 方法

Scheduleprocess 方法去找一条命令→Findonecommand，找到之后对它进行解析，创建工作流实例

交给 WorkflowExecuteThread 线程执行

WorkflowExecuteThread run 方法启动之后提交一个节点

放到优先级队列里面，构造处理的一些参数，通过 Netty 分发

Worker 接收任务并执行

• NettyClientHandler.java → channelRead() → processReceived() → TaskExecuteProcessor.java → process() → workerExecuteQueue.offer()

• WorkerManagerThread.java → workerExecuteQueue.take() 

• TaskExecuteThread.java → build taskRequest → taskChannel.create(taskRequest) →AbstractTask init() → handle()

通过 taskExcutethread 执行

通过代码跟踪，我们发现是调用队列里面的 offer 方法

通过 channelRead 方法接收 Master 发过来的信息，然后放在队列里面

通过 while 循环不停的处理队列里的任务

任务执行的参数都放在 taskExcutiontext 里面

TaskChannel 主要识别脚本任务类型（Flink、Shell、Python 等脚本）

任务执行完之后，通过 ResponseCommand 收集完信息反馈给 Master

04 思考与总结

01 优化返回方式

我们看源码的过程当中发现，比如说在 DolphinSchedulerr-api 跟前端交互接口的时候，请求参数返回的时候，它没有用 bean 来进行处理，什么意思呢？就是说它参数比较混乱，比如说自己的一些恶应用要接入 DolphinScheduler，或者说后续社区要提供一些 SDK，有 bean 的方式进行参数传递是比较好的。

02 表格式统一

还有一个在调研的过程当中，这个表的命名还有优化空间，表的命名格式有一些不同，比如说一些关系表，它后面加了 relation，然后有一些是加在前面的。

03 命名不友好

在看源码的过程当中，发现它的一些命名是不太友好的，如上图

05 拥抱开源

你如果从事软件开发，你有可能会想花时间去打造一下属于你自己的 IP，比如说你负责具体某个组件，成为核心开发者。

01 我们为什么要去学习开源？

1、提升技术功底：学习源码里的优秀设计思想，比如疑难问题的解决思路，一些优秀的设计模式，整体提升自己的技术功底，比如说在 DS 里面，用了大量的线程去处理一些 Event，然后使用了去中心化，还有使用队列，优先级队列等等

2、深度掌握技术框架：源码看多了，对于一个新技术或框架的掌握速度会有大幅提升，比如说你要去学习 DS 或者对它进行一个二次开发，你肯定要去了解一下其他知识，比如说 Netty，还有 Zookeeper

3、快速定位线上问题：遇到线上问题，特别是框架源码里的问题(比如 bug)，能够快速定位

4、拥抱开源社区：参与到开源项目的研发，结识更多大牛，积累更多优质人脉看源码

02 在开源社区的学习方法

• 先使用：先看官方文档快速掌握框架的基本使用

  
  

• 抓主线：找一个 demo 入手，顺藤摸瓜快速看一遍框架的主线源码，画出源码主流程图，切勿一开始就陷入源码的细枝末节，否则会把自己绕晕，凭经验猜

  
  

• 画图做笔记：总结框架的一些核心功能点，从这些功能点入手深入到源码的细节，边看源码边画源码走向图，并对关键源码的理解做笔记，把源码里的闪光点都记录下来，后续借鉴到工作项目中，理解能力强的可以直接看静态源码，也可以边看源码边 debug 源码执行过程，观察一些关键变量的值

  
  

• 整合总结：所有功能点的源码都分析完后，回到主流程图再梳理一遍，争取把自己画的所有图都在脑袋里做一个整合

参与贡献

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

参与 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) ，手把手教会您( 贡献者不分水平高低，有问必答，关键是有一颗愿意贡献的心 )。

添加小助手微信时请说明想参与贡献。

来吧，开源社区非常期待您的参与。