DolphinScheduler——奇富科技的调度实践

一、技术架构

二、业务挑战

2.1 调度任务量大

2.2 运维复杂

2.3 SLA要求高

三、调度优化实践

3.1 重复调度

3.2 漏调度

3.3 Worker服务卡死

3.4 任务重复运行

四、服务监控

4.1 方法耗时监控

4.2 任务调度链路监控

五、用户收益

原文大佬的这篇调度系统案例有借鉴意义，这里直接摘抄下来用作学习和知识沉淀。

一、技术架构

在我们公司的大数据离线任务调度架构中，调度平台处于中间层，通过数据集成平台、数据开发平台将工作流提交给调度平台。

二、业务挑战

2.1 调度任务量大

目前每天调度的工作流实例在3万多，任务实例在14万多。每天调度的任务量非常庞大，要保障这么多任务实例稳定、无延迟运行，是一个非常大的挑战。

2.2 运维复杂

因为每天调度的任务实例非常多，经历了几次调度机器扩容阶段，目前2个调度集群有6台Master、34台Worker机器。

2.3 SLA要求高

由于业务的金融属性，如果调度服务稳定性出问题，导致任务重复调度、漏调度或者异常，损失影响会非常大。

三、调度优化实践

我们在过去一年，对于调度服务稳定，我们做了如下2个方向的优化。第一，调度服务稳定性优化。第二、调度服务监控。

3.1 重复调度

用户大规模迁移工作流时，遇到了工作流重复调度问题。现象是同一个工作流会在同一个集群同一时间，生成2个工作流实例。经过排查，基于从A项目迁移到B项目的需求，在工作流上线时，用户通过提交工单，修改了调度数据库中工作流的项目ID，进行迁移。这么做会导致该工作流所对应的quartz（分布式调度器）元数据产生2条数据，进而导致该工作流重复调度。如图3所示，JOB_NAME为’job_1270’的记录，有2条数据，而JOB_GROUP不一样。查询源码job_name对应工作流的定时器ID，JOB_GROUP对应项目ID。因此修改工作流对应的项目ID，会导致quartz数据重复和重复调度。正确迁移工作流项目的方式是，先下线工作流，然后再修改项目ID。

SELECT count(1)FROM     (SELECT TRIGGER_NAME,        count(1) AS num    FROM QRTZ_TRIGGERS    GROUP BY  TRIGGER_NAME    HAVING num > 1 )t

3.2 漏调度

凌晨2点调度太集中，有些工作流发生漏调度。因此优化了quartz参数，将org.quartz.jobStore.misfireThreshold从60000调整为600000。

如何监控和避免此问题，监控sql摘要如下：

select TRIGGER_NAME,NEXT_FIRE_TIME ,PREV_FIRE_TIME,NEXT_FIRE_TIME-PREV_FIRE_TIME  from QRTZ_TRIGGERS  where  NEXT_FIRE_TIME-PREV_FIRE_TIME=86400000*2

sql逻辑是：根据quartz（分布式调度器）的元数据表QRTZ_TRIGGERS的上一次调度时间PREV_FIRE_TIME和下一次调度时间NEXT_FIRE_TIME的差值进行监控。如果差值为24小时就正常，如果差值为48小时，就说明出现了漏调度。

如果已经发生了漏调度如何紧急处理？我们实现了漏调度补数逻辑通过自定义工作流进行http接口调用。如果监控到发生了漏调度情况，可以立即运行此工作流，就能把漏调度的工作流立即调度运行起来。

3.3 Worker服务卡死

这个现象是凌晨调度Worker所在机器内存占用飙升至90%多，服务卡死。

思考产生该问题的原因是，调度worker判断本机剩余内存时，有漏洞。假设设置了worker服务剩余内存为25G时，才不进行任务调度。但是，当worker本机剩余内存为26G时，服务判断本机剩余内存未达到限制条件，那么开始从zk队列中抓取任务，每次抓取10个。而每个spark的driver占用2G内存，那么本地抓取的10个任务在未来的内存占用为20G。我们可以简单计算得出本机剩余内存为26G-20G为6G，也就是说抓取了10个任务，未来的剩余内存可能为6G，会面临严重不足。

为了解决这个问题，我们参考Yarn，提出了”预申请”机制。预申请的机制是，判断本机剩余内存时，会减去抓取任务的内存，而不是简单判断本机剩余内存。

如何获取将要抓取任务的内存大小呢？有2种方式，第一种是在创建工作流时指定本任务driver占用的内存，第二种是给一个固定平均值。

综合考虑，采用了第二种方式，因为这种方式对于用户来说，是没有感知的。我们对要抓取的每个任务配置1.5G（经验值）内存，以及达到1.5G内存所需要的时间为180秒，抓取任务后，会放入缓存中，缓存过期时间为180（经验值）秒。剩余内存计算公式，本机剩余内存 = 【本机真实物理剩余内存】—【缓存中任务个数*1.5G】

还是同样的场景，本机配置的剩余内存为25G，本机实际剩余内存为26G，要抓取的任务为10个。每个任务未来占用的driver内存为1.5G。简单计算一下，本机剩余内存=26G-10*1.5G。在“预申请”机制下，本机剩余内存为1G，小于25G，不会抓取，也就不会导致Worker机器的内存占用过高。那么会不会导致Worker服务内存使用率过低呢，比如shell、python、DataX等占用内存低的任务。结论是不会，因为我们有180秒过期机制，过期后，计算得到的本机剩余内存为变高。

实施上文的内存预申请机制后，最近半年没有遇到由于内存占用过高导致worker服务卡死的问题。以下是我们加上内存预申请机制后，worker内存使用率情况，可以看见worker最大内存使用率始终稳定保持在80%以下。

3.4 任务重复运行

在worker服务卡死时，我们发现yarn上的任务没有被杀死，而master容错时导致任务被重复提交到yarn上，最终导致用户的数据异常。

我们分析后发现，任务实例有一个app_link字段，该字段存放用户提交的yarn任务的app id，而第一次调度的任务的app id为空。排查代码发现worker在运行任务时，只有完成的yarn 任务，才会更新app_link字段。这样导致master在容错时，拿不到app id，导致旧任务没有被杀死，最终导致任务重复提交。

我们进行的第一个改进点为，在worker运行yarn任务时，从log中实时过滤出app id，然后每隔5秒将app id更新到app_link字段中。这样yarn任务在运行时，也就能获取到app id，master容错时就能杀死旧任务。

第二个改进点为，在worker服务卡死从而自杀时，杀死本机上正在运行的调度服务，这样可能master就不需要进行容错了。实施改进点后，最近半年没有遇到重复调度的yarn任务了。

四、服务监控

一个稳定的系统，除了代码上的优化，一定离不开完善的监控。DolphinScheduler 对外提供了 Prometheus 格式的基础指标，我们新增了一些高优指标，并集成到公司内部的监控系统。通过监控大盘来查看调度系统的健康状况，并针对不同级别的prometheus指标和阈值，配置电话 / 钉钉报警。

4.1 方法耗时监控

我们通过byte-buddy、micrometer等，实现了自定义轻量级java agent框架。这个框架实现的目标是监控java方法的最大耗时、平均耗时、qps、服务的jvm健康状况等。并把这些监控指标通过http暴露出来，通过prometheus抓取，再通过grafana进行可视化展示，并针对不同级别的prometheus指标和阈值，配置电话 / 钉钉报警。

例如以下是master访问zk和quartz的最大耗时，平均耗时，qps等。