作业帮基于 Apache DolphinScheduler 3_0_0 的缺陷修复与优化

文|作业帮大数据团队（阮文俊、孙建业）
背 景

基于 Apache DolphinScheduler （以下简称DolphinScheduler）搭建的 UDA 任务调度平台有效支撑了公司的业务数据开发需求，处理着日均百万级别的任务量。
整个 UDA 的架构如下图所示，其中我们的引擎层主要基于 DolphinScheduler 的 Master 和 Worker 服务搭建，充分使用了 Master 节点良好的高并发调度能力、worker 节点的分组隔离能力。我们通过一层适配层对 DolphinScheduler 对外暴露的接口进行了封装和增强来适配我们的业务场景。

随着使用的深入，我们发现 DolphinScheduler3.0.0 版本中存在一些难以通过运维本领规避的题目，这些题目影响数据平台的稳定性，导致隔一段时间必要重启服务来使集群恢复正常，并且核心组件对外暴露的可观测性指标十分有限，导致题目的排查定位过程十分繁琐。我们以为一个可以稳定运行的调度引擎应该具备以下能力

• 反压与过载保护机制：当体系负载比较高的时候，能够自动推迟任务执行，以保护自己。
  
• 任务分配均衡：保证任务分配的均衡性，避免任务倾斜题目。
  
• 无副作用运行：调度引擎在运行过程中不应对自身产生“副作用”，确保能够持续的长时间稳定运行。
  
• 集群可观测性：具备全面的可观测性指标，能够通过这些指标评估集群的健康状态。
  

在这些能力上，开源版 3.0.0 的 Apache DolphinScheduler 尚存在一些题目，对此，我们进行了系列优化改造和修复，同时积累了丰富的运维履历。
优化实践

2.1 题目修复

2.1.1 HadoopUtils 引发的线程泄漏题目

在某次巡检的过程中，我们发现服务节点的线程数在过去一段时间呈显着的上升态势，根据履历判断，应该是程序中存在线程泄漏的地方，联合 metrics 发现泄漏速率为恒定速率，并且与任务并发量无关。

通过堆栈发现泄漏的线程主要是与 HDFS 相关，进一步将代码范围缩小至 HadoopUtils 之后，我们发现此处存在引发线程泄漏的代码逻辑。在 DS HadoopUtils 中存在一个 Cache，会以恒定的速率不断生成新的 HadoopUtils 实例，并放入 Cache，在 HadoopUtils 实例化的过程中会创建 HDFS FileSystem，但是却不会关闭原有的 FileSystem。
开源的 DS 在使用 HDFS FileSystem 时泄漏速率比较慢，不易发现。我们在生产环境中使用的是腾讯云提供的 CosFileSystem 插件，该插件中会使用多线程来加快文件的上传 / 下载操作。通过插件中的线程数目比对，与我们泄漏的线程数完全一致。

至此，我们确定线程泄漏的原因是 HadoopUtils 在更新 Cache 的时候没有关闭 FileSystem，于是我们在更新 Cache 的时候关闭 FileSystem，并通过读写锁保证不会由于异步关闭导致文件操作失败，乐成的办理了线程泄漏的题目。对比上线前后的 JVM 线程指标，修复之后线程数保持在一个小恒定范围内。

2.1.2 TaskExecuteRunnable 内存泄漏引发 CPU 飙高题目

在生产环境中，我们发现在任务量没有发生显着变化时，Master 服务随着运行时间越长，其 CPU 使用率出现增长趋势。

通过分析火焰图，我们发现 Master 中存在一处代码逻辑，随着程序的运行时间越久，这段代码逻辑对 CPU 的斲丧会越来越高。

通过梳理代码逻辑，我们发现 DolphinScheduler 的 Master 服务在运行工作流时采用事件驱动的方式，每个任务实例在运行过程中会生成一个对应的 TaskExecuteRunnable 对象，任务在运行过程中产生的生命周期事件会存放在 TaskExecuteRunnable 对象中。会有一个后台线程轮询当前服务中存活的 TaskExecuteRunnable 对象，然后提交事件处理任务到事件线程池。
不外，当任务运行结束之后，TaskExecuteRunnable 并不会被释放，还会存放在任务事件线程池中，这就会导致任务事件处理线程空转时间越来越长。
通过分析堆栈，我们的判断得到了验证，TaskExecuteRunnable 的确会泄漏，不外由于 TaskExecuteRunnable 占用内存很少，因此很难从内存中反应出来。我们的集群中有 4 台 Master，任务实例数一天百万左右，因此对于单台 Master 一天会泄漏约莫 25w TaskExecuteRunnable，随着时间的积累会拖慢引擎的事件处理。

于是我们进行了代码修复，在任务执行结束之后，移除内存中的 TaskExecuteRunnable 对象。对比修复前后的 JVM CPU 指标，修复之后，Master 的 CPU 指标随着运行时间始终维持在一个小的恒定范围内。

2.1.3 Master 执行逻辑任务重复提交事件，导致事件堆积题目

DolphinScheduler 中任务分为两类，分别为以 dependent 为代表的逻辑任务和以 Shell 为代表带的物理任务，其中逻辑任务在 Master 中执行，物理任务在 Worker 中执行。不管是逻辑任务还是物理任务在 Master 处理过程中都会履历以下阶段。

逻辑任务和物理任务的区别在于 Dispatch 和 Run 阶段的实现差别，对于逻辑任务不必要触发真正的 Dispatch，Run 阶段运行在 Master 中。而物理任务在 Dispatch 阶段会将任务分发给 Worker，Run 阶段运行在 Worker 中。
无论是哪种任务，当 Dispatch 阶段执行乐成之后，会注册到 StateWheelThread 中，该组件会定时的每隔 5 秒钟为每个任务生成一个 TaskStateChangeEvent，提交到任务的事件队列中，TaskStateChangeEvent 被处理的时候会触发 Run 任务，对于逻辑任务会不断的通过 TaskStateChangeEvent 触发执行。

这里是一个典型的生产消费模型，生产者以固定的速率（每隔 5 秒）生成事件写入队列，消费者异步的从队列中消费事件。
因此当消费者处理的速率小于生产者生产的速率时，这里就会出现事件堆积。
而现实情况下，由于生产者生产事件的时候是纯内存盘算，没有任何 io 壅闭，而消费者处理事件的时候必要多次查询 db。对于 Dependent 这类逻辑任务的运行时间通常都很长，因此假如达到肯定的并发量，这里极大概率会出现事件积压，导致整个 master 中所有任务的状态事件处理出现耽误、增加数据库压力，严峻的话还会导致 Master OOM 服务宕机。
在我们的测试环境中，单台 Master 服务，事件线程池大小为 100，Dependent 并发数超过 500，此时就 Master 中的 StateEvent 就会出现堆积的情况。

我们发现堆积的事件都是用来触发逻辑任务 Run 阶段，并且对于同一任务实例存在多个重复的触发事件，我们通过对事件去重从而修复堆积的情况。在修复之后，事件的堆积情况得到办理，一旦事件的消费速率低于增长速率，事件的堆积量最多为任务的并发数，不会出现不停积累的情况。

2.1.4 Master 任务调度不均匀

Master 在分配任务给 Worker 的时候，会使用负载均衡策略，使任务的分配只管均衡。默认的均衡策略是 LOW_WEIGHT，该策略会通过 Worker 的心跳信息来盘算一个负载量，会将任务分配给负载量最低的 worker。

在现实的使用过程中我们发现在大多数情况下，这种负载策略会出现严峻的任务分配不均衡的情况，在同一个 WorkerGroup 下，差别的 Worker 被分配到的任务量可能会相差几十倍。

究其原因我们发现主要是由两个方面导致

• 负载盘算的值无法代表负载情况
  

在盘算 Worker 节点的负载时，Master 会对 Worker 的 CPU、内存、Load、等待任务数分别加一个权重来做归一化，但是针对各资源加权值和归一化算法表达不严谨。导致盘算出来的负载值现实上并不能正确的反应 Worker 的真实负载情况，并且现实生产很难通过调节权重得到一个真实的值。

• 负载盘算不是实时的
  

Worker 的心跳上报是定时上报，Master 在分发任务时使用的 Worker 心跳数据并不能反映当前 Worker 的真实情况，这会导致某个时刻一旦出现一个负载量偏低的 worker，master 在接下来一段时间中可能会将大量的任务都发送给这台 worker，从而导致任务倾斜。
分析完原因之后，我们决定使用 RANDOM 策略来分发任务，保证 Master 在分发时绝对均衡，然后由 Worker 自己通过自身负载决定是否要接受 Master 的分发请求。对比修复前后同一个 worker 分组下差别 Pod 接收的任务，发现修复后差别的 pod 接收的任务变得均衡，不再出现任务倾斜的情况。

2.1.5 Master 事件处理卡住题目

在生产环境中，我们发现 Master 的 CPU 持续升高，通过服务日志发现 Master 不停在处理某个事件，并且伴随异常，我们猜测此时出现了事件死循环的情况。通过研究 DolphinScheduler3.0.0 中 Master 事件驱动流程我们发现在该版本中存在三类事件。

• WorkflowStartEvent
  

WorkflowStartEvent 是工作流启动事件，该事件是由一个单独的后台线程产生，并且由一个单独的后台线程处理，用于启动工作流，工作流的元数据出现异常时会导致 WorkflowStartEvent 执行出现异常，此时异常的 WorkflowStartEvent 会不停重试并壅闭后面其他事件，直到在数据库中对元数据进行修复。

• StateEvent
  

StateEvent 是工作流和任务执行相关的事件，用于驱动 DAG 拓扑执行。
StateEvent 有以下事件类型：工作流状态变更、任务状态变更、工作流超时、任务超时、TaskGroup 中的任务被叫醒、任务重试、工作流壅闭。
对于一个工作流来说，里面所有的 StateEvent 都存储在一个队列中，事件按照进入队列的先后顺序被执行。并且采用的是 DFS 的方式被线程池消费，即一个队列被 fire 的时候会被分配给一个线程，该线程直到处理完队列中的所有事件才会退出，假如一旦有某个队列在处理时无法退出，那么线程会被不停占用。

• TaskEvent
  

当任务实例在运行时发生了变化会生成 TaskEvent，即该事件是由 Worker 发送的任务数据所转换而来，以下情况都会生成 TaskEvent，TaskEvent 处理流程和 StateEvent 类似。

• 任务实例被分发乐成了，那么会触发 Dispatch 类型的 TaskEvent。
  
• 任务实例耽误执行了，那么会触发 Delay 类型的 TaskEvent
  
• 任务实例开始运行了，那么会触发 Running 的 TaskEvent
  
• 任务实例运行结束了，会触发 Result 类型的 TaskEvent
  

值得注意的是以上三种事件都采用死信队列的方式存放，即只有当事件被处理乐成才会将事件从队列移除，社区最初这么计划是希望在某些情况下由于基础办法故障，例如 db 抖动等不会影响到事件的处理，但现实上有很多其他的意外情况会导致事件处理失败，例如数据库存在非正常数据，事件发送过程中出现乱序等。
我们以为对于引擎来说，必要避免由于某一个工作流事件处理出现题目，从而影响到引擎的稳定性。因此，我们移除了这里的死信队列，当事件处理失败的时候，会直接抛弃事件，并将工作流快速置为失败，由上层进行重试，并联合 Metrics 监控各类事件的处理情况。修复后，Master CPU 保持稳定，服务日志也不再出现不停重复处理某个事件。

2.2 稳定性优化

2.2.1 工作流实例健康检查

现在 DolphinScheduler 中 Master 执行工作流的时候会将工作流实例的元数据存储在内存，然后通过事件驱动的方式去进行状态流转，直至工作流中所有的任务都结束然后将工作流实例从内存卸载。在某些情况，例如网络原因导致事件丢失，或者事件在处理过程中由于状态机 bug 处理失败从而丢失，此时会导致工作流实例处理流程卡住，从而导致工作流实例成为孤儿实例，即永久不可能结束。
此时假如发现了可以在上层通过 kill 的方式去停止工作流实例，从而卸载，不外这种方式存在两个题目。一是依赖业务方自行检测，必要业务方定期的巡检整个体系，当业务方发现题目时往往业务已经受到了影响。二是处理的方式很繁琐，一旦运行的工作流实例数比较多的时候，逐个操作本钱比较高。我们希望调度引擎能够有自检功能，能够自己检测工作流实例是否已经变成僵尸实例，并且自动上报，自动做恢复操作。
对此，我们进行了优化，在 Master 中添加一个组件 WorkflowInstanceHealthCoordinator，该组件用于定期对当前 Master 运行中的每个工作流实例执行健康检查。在健康检查的时候会通过 HeartbeatEmitter 去触发工作流实例的心跳检测，当一连多次的心跳检测失败之后，会通过 DeadWorkflowInstanceHandler 去清除该工作流实例，并上报 metrics。

整个检测主要是由 WorkflowInstanceHealthCoordinator 负责，该组件在每个 Master 中采用单例的形式，里面包含一个后台线程，和 EventExecuteService 工作模式类似，当一个工作流实例被加载到 Cache 之后，会同时在 WorkflowInstanceHealthCoordinator 中注册自己，当工作流实例执行结束从 Cache 中移除的时候也会同步从 WorkflowInstanceHealthCoordinator 移除。

WorkflowInstanceHealthCoordinator 中有一个后台线程会定期的（默认 5min 一次，可自定义设置检测间隔）对注册进来的工作流实例做健康检查。
健康检查的方式是通过对工作流实例中所有未结束的任务做心跳探测，假如探测乐成，则表明该工作流实例是存活的，假如探测失败，则表明该工作流实例可能已经出现了异常。对一个工作流实例假如探测失败的次数超过了阈值，我们以为该工作流实例已经成为僵尸工作流实例，我们现在会进行告警，由运维同学介入，当前我们尚未实现自动故障恢复，由于此类僵尸实例发生的情况不会很多，后续我们会考虑实现对僵尸实例自动运维。
2.2.1 Worker 中任务事件 TTL

DolphinScheduler 中 Worker 主要职责是接受任务，执行任务，上报任务事件。
其中任务事件在上报的时候存放在内存中的一个死信队列中。

整个过程为

• 任务执行过程中生成任务事件，并将事件提交到死信队列，每个任务会有一个单独的死信队列
  
• Worker 中有一个后台线程会定期轮询死信队列，当事件达到重试间隔之后会重新发送事件给 master
  
• Master 在处理完事件之后会发送对应事件 ACK 给 worker，worker 收到 ACK 之后会清除事件，当死信队列中所有事件都为空，并且任务执行结束，此时会卸载死信队列。
  

这样做的好处是能够避免由于网络抖动或者 master 由于故障而导致某段时间内事件上报不乐成从而丢失事件，不外这样也会导致可能出现内存泄漏的题目。例如，假如 master 发生容错，那么会进行工作流容错，容错的时候会先 kill 任务，然后重新提交，在 kill 的时候假如发送 RPC 给 worker 失败了，此时 worker 中的任务事件将永久不会被清除，并且由于工作流实例发生了容错，此时某些任务事件可能无法发送给容错后的 master，即会不停重试，变成僵尸消息。
即一旦消息的目的地发生了变化，但是 worker 感知不到，那么会导致消息泄漏到内存。
一旦发生泄漏，可能会导致重试线程中堆积大量的无效事件，这会占用线程资源，导致有效事件发送出现耽误，并且这类无效事件永久不会被释放，会造成内存泄漏，影响服务稳定性。
对此，我们在事件中添加了 TTL，每个事件在创建的时候会带有 createTime，假如 currentTime-createTime>ttl，那么表明事件在给定的时间内没有发送乐成，此时阐明事件可能已经出现了泄漏，会在 prometheus 中打点，并自动从死信队列队列中清除。

2.2.2 汗青数据保留策略

随着体系使用时间的增长，数据库和磁盘数据逐渐积累，会影响服务运行和数据库稳定性。现实运维中我们发现，数据库中增长的主要是一些实例元数据，这些数据的积累会导致数据库压力越来越大，同时会伴随慢查询越来越多。磁盘中增长的主要是任务实例日志和任务实例工作目录，这些数据的积累会导致磁盘可用容量和 inode 变得越来越少。
我们希望程序能够自动的清理无用的汗青数据。例如，在数据库中仅保留最近一个月的运行实例数据。磁盘上保留最近一周的临时文件，超出保留期限的数据则自动删除，以淘汰人工运维的工作。
由于 DolphinScheduler 原生的删除接口在做数据清理的时候是按照工作流实例的维度，即清理汗青数据的时候必要先找出工作流实例下的任务实例，然后分别清理每个任务实例的数据，这个过程涉及大量的数据库操作和 RPC 操作，并且执行批量删除操作的时候会给服务带来很大的压力，不适用大批量数据的清理。为此，我们分别在 Master 和 Worker 中添加了 InstanceDBPurgerThread 和 TaskFilePurgerThread 两个后台线程组件，分别负责数据库实例数据和磁盘文件的定期检查、上报与清除工作。
数据库方面主要清理工作流实例、任务实例和告警事件等数据。磁盘方面，主要是清理 exec 目录下的临时文件夹，log 目录下的任务实例日志等数据。同时，通过暴露相关的 metrics 对这些数据量进行监控。将数据库和磁盘的清理分开可以极大的加快汗青数据的清除速率。

2.3 巡检流程

为了确保体系稳定运行，我们不但设置了大量告警，还定期进行日常、周、月巡检。通过巡检，我们能够提前发现潜在题目，不断的美满自动化运维流程。我们现在发现的大多数题目都是通过巡检提前发现，避免了对业务造成现实影响。
同时，巡检也帮助我们不断优化我们的监控大盘和告警项。现在我们从集群、项目、WorkerGroup 和 Pod 等维度搭建了监控面板和告警，以辅助巡检工作。

在日常巡检中，我们主要关注集群、项目、WorkerGroup 三个维度下的指标。在集群维度上，关注 Master Slot 变化、集群水位、并发量、资源使用等稳定性指标。在项目和 WorkerGroup 维度，关注异常任务、任务量的同比变化，WorkerGroup 下 Slot 使用率及业务运行情况。Pod 维度则用于周巡检和题目排查。
将来规划
现在 DolphinScheduler3.0.0 已经在我们的生产环境中稳定运行，我们针对使用场景中发现的题目，在不进行大规模架构调整的条件下做出了修复和优化，并且沉淀出了一套适用于当前业务场景的运维本领。社区在后续版本中对某些题目进行了更美满的修复，如针对逻辑任务事件壅闭的题目，重构了整个逻辑任务执行流程；针对状态机卡住题目，重构了状态机模型等。将来，随着业务量和使用场景的扩展，我们会考虑版本升级到 3.2+ 版本，以尽可能与社区保持同步，并将我们所做的一些优化项反馈至社区。

本文由 白鲸开源 提供发布支持！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。