关于spark:BIGO-如何做到夜间同时运行-24K-个工作流实例

点亮 ⭐️ Star · 照亮开源之路

GitHub:https://github.com/apache/dol…

​

精彩回顾

近期，BIGO 的大数据研发工程师许名勇在社区线上 Meetup 上给大家分享了主题为《DS 及 SPARK 在 BIGO 的利用和改良》的演讲。

次要介绍了 BIGO 如何应用 DophinScheduler 来调度以 SPARK 为主的多种类型的离线工作，以及为了满足业务需要、晋升用户应用体验，在 DS 和 Spark 上所做的各种改良。

讲师介绍

​

许名勇

Bigo 大数据研发工程师

文章整顿：白鲸开源 - 曾辉

明天的演讲会围绕上面三点开展：

1. Apache DolphinScheduler 利用详情
2. Apache DolphinScheduler 改良
3. Apache Spark 改良

DS 利用详情

01 为什么抉择 DS？

咱们原来的调度平台用过很多个，无一例外都碰到一些难以满足本身需要的痛点问题：

• Oozie :查看日志不便，短少工作状态统计、工作流数量多了后存在调度压力等;
• Airflow : 须要用 python 代码来绘制 DAG，存在肯定应用门槛;
• Crontab：就更原始了，而且是单点，应用不便。

通过调研后，咱们最初抉择 DS 调度沿用至今，因为在肯定层面上 DS 满足咱们所有的需要。

对于咱们团队外部来说，次要的劣势是：

1. 能对工作流 DAG 可视化进行编辑，简略易用，日志查看也比拟不便。
2. 去中心化的多 master 多 worker 架构，能够线性扩大，从而保障了高可用。
3. 反对的工作类型很丰盛，符合大数据生态，不便定制和革新。
4. 还反对补数，这个性能很实用。

当然，抉择 Apache DolphinScheduler 还有一个契机，咱们正在自研一站式的数据开发平台，须要一个新的工作流调度零碎，综合所有的因素最终促成了咱们抉择海豚调度。

02 DS 集群详情

DS 在咱们生产环境运行有 1 年工夫了，目前部署的状况是：1 台 alert 服务，3 台 api 服务，10 台 master 9 台 worker，这些服务混合部署在 10 台物理机上。

​

革新后的 DS 工作根本都是提交到 Yarn 上运行, Yarn 集群规模目前是有 2000 多台 node manager 节点，集群总共有 500 多 T 内存、10多万 core。

​

​

​

03 DS 作业详情

对于作业状况，目前曾经实现了最次要的 Oozie 工作流迁徙，当初日均调度 1.8W 多个工作流实例，5W 多个工作实例。

​

在夜里顶峰时段同时有 2400 多个工作流实例运行，这些实例次要以 Spark、Sparksql、Shell、Python、Sqoop 节点为主。

2 DophinScheduler 改良

01 用户体验晋升

为了晋升用户体验，咱们围绕升高开发成本，简化工作配置做了很多改良。

买通 OA，免去注册。不便用户登录。对于这种公司外部零碎来说，还须要注册显然是不太不便的。

而后在 DAG 编辑页面实现 开发、上线、树形图查看、运行、定时治理等一站式操作，尽量让用户在一个页面就能实现工作流的各个操作。

​

在 树形图页面聚合 了单个工作流定义的所有实例列表，这样能够更加清晰的看到属于当前工作流的实例。

首页提供定时调度详情。展现调度的实例胜利与否的状态，这样用户对本人的定时调度最近运行的状况就能够高深莫测，

减少页面分层级树形导航条，提供在我的项目、工作流、树形图页面疾速切换等等。

因为用户是以我的项目为粒度组织工作流，一个用户可能会有多个我的项目，这个疾速切换能够不便用户从一个我的项目下的工作流疾速切到另一个我的项目下的工作流，进步用户的开发效率。对于工作流的体验改良咱们还做了很多，因为篇幅无限，我就不开展来讲。如果您对此有趣味，欢送来社区跟我探讨。

在依赖抉择方面，简化了依赖节点的配置。

原先配置一个依赖节点，是要先抉择我的项目，再抉择工作流，用户反馈应用起来不太不便，因而咱们做了相干改良，在配置依赖节点的时候能够间接按 工作流 ID、工作流名称或工作流 owner 来搜寻依赖工作流进行配置。

搜寻配图

咱们还在首页的详情页面，提供了 便捷查看上游依赖实例详情 的入口，让用户一眼就能看到当前工作流实例的上游是哪个实例没有实现。

​

这样能够在呈现工作提早的时候不便排查是不是上游还没实现导致的。

除此之外，还新增了一张表来记录工作流的上下游关系，在保留和更新工作流定义的时候去更新这个依赖表，而后提供了工作流依赖关系查看性能，以树形图的模式展现一个工作流的所有上游或是上游。

​

咱们还反对了 上游依赖的批量替换。比方，一个工作流可能有很多上游，这时因为业务需要可能心愿用新的工作流替换原先这个工作流，就能够用这个性能同时把上游依赖也替换掉。

​

在补数方面，咱们做了比拟大的改良, 其中最常见的业务场景，就是能够反对触发上游，用户在补数的时候可能会心愿当前工作流补完了，带动上游工作流也一起补。

​

那这种状况下补数去触发上游，其实也是一个工作流实例的 DAG 了，因而咱们不仅反对查看补数的进度，还能够晓得以后补数是补到哪个工作流的哪个调度工夫。

​

限度同一个调度工夫的实例同时运行。这是因为如果一个上游有多个上游开始触发上游补数时，这个上游可能会补数补重，也就是同时有两个雷同调度工夫的补数实例在跑，当然用户如果误补数的话也有这种状况产生，所以咱们做了限度来防止同时跑。

告警治理，咱们把超时告警和失败告警整合到了定时调度配置页面来一起设置，并反对告警到值班组。因为咱们理论需要是只须要对定时调度的实例去告警，其余手动运行或是补数的实例不须要告警，因而这两个超时和失败告警整合到定时调度配置页面，在配置调度的时候去配置告警。

​

告警人能够填值班组，而不仅仅是用户，值班组里的成员每天都会排班。

超时和失败告警咱们改成了用户如果没解决会每小时发一次告警，原先是发送一次就没了，这有可能会让用户脱漏，导致工作流没有失去及时处理。

如果用户不须要解决这个实例或者单纯想勾销告警，咱们也在实例上提供了勾销告警的抉择。用户也能够在工作流上屏蔽告警，并且能够同时屏蔽上游。在屏蔽工夫范畴内将不会发送告警。

最初还提供了 工作流定时调度到期揭示性能。如果工作流行将到期，将会发送告诉给用户，前端页面也会减少醒目的提醒揭示用户工作流快过期了，要及时延期，防止影响工作流的失常调度。

02 零碎层面改良

Worker 反对工作无需重跑的 failover。以前 Worker 重启，Worker 上正在运行的工作须要 Kill 从新提交。

这就有几个问题：

1. 工作运行工夫比拟长，间接 kill 重跑代价大；
2. 重跑比拟节约集群资源；
3. 限度 worker 同时运行工作数量的下限；

咱们对此的优化是 革新工作提交形式，不论是 shell，还是 python，java 都提交到 yarn 下来执行，不在 Worker 本地执行。

对于 spark 和 spark-sql 工作，以 cluster 模式提交到 yarn 上，client 提交实现后过程立马完结，不在 Worker 上驻留。

同时在工作实例里记录提交到 yarn 上后的 app id 信息，提交胜利后开启异步工作 track 在 yarn 上的执行状态。

如果 Worker 宕机，容错后能够由其余 Worker 接管这个工作，持续 track 在 yarn 上的运行状态。这样就实现了无需工作重跑的 failover。

Worker 工作只提交一次。咱们目前应用的还是 1.3.8 版本，容错的时候有个比拟大的问题，就是会把 master/worker 启动后的实例也容错，这会导致工作反复提交，因为咱们是把工作提交到 yarn 上执行，反复提交体现就是：同一个工作实例同一时间会在 yarn 上有多个作业在执行。

这会导致工作流跑的数据不对，另一个就是会重大梗塞集群，影响其余工作流的失常运行。

针对这个问题，咱们做了绝对应的措施。

1. 在实例表减少一个字段，master/worker 容错时依据字段来来判断实例是否须要容错。
2. 是在 master 端重试散发 task 时，优先散发到上一次散发的 worker。
3. 提交到 yarn 的 task 设置 yarn tags，failover 后的 task 从新散发到 worker 后如果没有 app id 信息，则通过 yarn tags 查看 yarn 上该 task 是否曾经提交或者在运行了。

通过这些措施来保障 worker 提交工作只提交一次。

升高对数据库的压力。目前应用的这个版本 master 对线程的应用过重，对数据库有比拟大的压力。每个工作流实例和工作实例都独占一个线程，状态轮询存在很多数据库 IO 操作。

另外一个是依赖节点，比方存在 24 小时依赖类型，这种依赖类型每个小时实例须要查一次状态，一个依赖项就要查 24 次数据库，如果这种依赖节点比拟多，又是集中在某几个时刻运行，那就有可能 并发查问的量过大 ，导致数据库沉闷 连接数过高，甚至打满数据库连接池，造成重大的结果。

咱们对此做的优化是：

1. 增大工作流实例和工作实例状态轮询的距离；
2. 实例表减少索引，放慢 sql 查问速度；
3. 依赖节点查问上游实例状态时通过随机化扩散查问，升高同时查问的并发数；
4. 减少缓存，缩小不必要的反复查问；

这些措施肯定水平上减少了依赖节点的执行工夫，但防止了数据库沉闷连接数过高的问题，晋升了零碎的稳定性。

反对 Spark 工作灰度

针对 Spark 节点和 sql 节点咱们是以 Spark-submit 或者 Spark-sql 形式提交的，如果要变更 Spark，那在对 Spark 工作灰度的时候存在几个问题：

1. 单个工作灰度能够增加 spark.yarn.archive 参数来指定 Spark 版本，但如果要批量灰度就不不便了;
2. 咱们也能够在 Worker 级别灰度，这样无奈指定具体的工作流，也不好管制灰度规模。

因而，咱们开发了灰度治理的性能。能够在前端操作，批量增加一批工作流到灰度列表，增加到灰度列表的工作流在 master 散发工作到 worker 时，利用上灰度信息。

并且能够依照工作类型和工作流 owner 筛选工作流来灰度，当工作流从灰度列表移除就示意勾销灰度。

这样，就能够比拟不便的对工作进行灰度降级。

3 Spark 改良

01 小文件合并

对于小文件合并，家喻户晓，小文件与 hdfs 是不敌对的，太多的小文件会给 hdfs 带来重大的性能瓶颈。

而 Spark 作业并行度高的话又容易产生小文件，因而小文件合并对于 Spark 来说是非常有必要的。

小文件合并的实现计划有很多，比方能够在最初减少一个 shuffle 来管制文件数量。

​

咱们的计划是：在文件 commit 过程中去实现的。首先看 Spark 工作的文件提交机制，在 driver 端会先 setupJob，而后在 executor 端执行 task 的时候 setupTask，创立 task 长期目录，在 commitTask 的时候将数据文件从 task 长期目录转移到 job 长期目录。

最初 driver 端执行 commitJob 办法，将各个 task 工作提交的数据文件，从 Job 长期目录转移到 Job 的最终目标目录。

小文件合并计划 是：在 driver 端 setup job 时，在 Job 最终目标目录下创立长期的 merge 目录，而后依照失常的流程将数据写到这个长期目录，在 driver commit job 之后，所有的数据文件都曾经转移到了这个长期目录中。

此时去计算这个长期目录下每个分区门路下的数据文件的均匀大小，如果只有一个文件或大于指定的阈值，就不须要合并，能够间接挪动到最终目标目录。

如果小于指定阈值，则依照分区门路下文件大小之和除以阈值来计算合并后的文件数量，而后将分区门路下的文件读成 HadoopRDD 或 FileScanRDD，再依照计算的合并后文件数量进行 coalesce 操作，而后启动一个 Spark Job 将 RDD 数据写到最终目标目录中。

这个小文件合并计划在咱们线上也利用了很长时间，尽管须要启动一个额定的 job，但依然带来了较大的收益。比方，它大大减少了文件数量，加重 NameNode 的负载压力。也升高了 Spark 作业 driver OOM 呈现的概率，进步了数据读取效率，放慢了执行速度。

02 AQE 优化

​

对于 AQE 的优化。AQE 是 Spark3 的重要性能，通过收集运行时的统计信息，来动静调整后续的执行打算。AQE 次要有三个优化场景：动静分区合并、动静调整 Join 策略以及动静优化歪斜 Join，实现原理就不在这里赘述了。

AQE 的进一步优化。

第一点，分区合并能够让 reducer 解决的数据量适中，但如果物理打算中存在 expand 算子，会导致数据收缩，即便 reducer 读取的数据量适中但通过数据收缩之后，也会造成极大性能倒退，这种状况不适宜进行分区合并或是须要调整分区合并的大小，针对这种状况，咱们抉择了针对这个 stage 不进行分区合并的操作。

​

​

​

第二点 ，目前歪斜 Join 优化是依据分区数据量的大小来判断是否存在歪斜的， 扩大歪斜 Join，能够反对收集分区的行数信息，这样既能够依据分区数据量大小，也能够依据分区的行数来判断一个分区是否歪斜。

第三点，咱们利用收集到的行数信息，在合并分区时还会依据数据量大小和行数的比值来决定是否进行分区的合并。

AQE 的优化咱们也上线了很长时间，能够说对大部分作业都缩短了运行工夫，进步了执行效率。通过统计，开启 AQE 优化后，咱们集群整体作业的 均匀运行工夫缩短了 10%。

明天的分享就到这里，如果大家对我文章内容有任何问题，能够在社区外面来找我交换，感激大家凝听和关注。

参加奉献

随着国内开源的迅猛崛起，Apache DolphinScheduler 社区迎来蓬勃发展，为了做更好用、易用的调度，真挚欢送酷爱开源的搭档退出到开源社区中来，为中国开源崛起献上一份本人的力量，让外乡开源走向寰球。

参加 DolphinScheduler 社区有十分多的参加奉献的形式，包含：

​

奉献第一个 PR(文档、代码) 咱们也心愿是简略的，第一个 PR 用于相熟提交的流程和社区合作以及感触社区的友好度。

社区汇总了以下适宜老手的问题列表：https://github.com/apache/dol…

非老手问题列表：https://github.com/apache/dol…

如何参加奉献链接：https://dolphinscheduler.apac…

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

参加开源能够近距离与各路高手切磋，迅速晋升本人的技能，如果您想参加奉献，咱们有个贡献者种子孵化群，能够增加社区小助手 Leonard-ds，手把手教会您(贡献者不分程度高下，有问必答，要害是有一颗违心奉献的心)。

增加小助手时请阐明想参加奉献，来吧，开源社区十分期待您的参加。