任务治理“三板斧“之任务编排

背景

先说说写这个系列的目的吧，有认识我的朋友知道，我在杭州某不知名电商工作，负责数据平台方向的开发，虽然我之前一直分享的都是某些组件的用法，但我的本职工作还是围绕着各个组件去打造去中心化的数据能力。我们团队从去年开始，一直在朝着这个方向努力；经过一年的积累，我们也做出了部分的产品，如元数据系统、开发平台、调度中心、数据交换等等；但随着我们的业务在不断的扩张与发展，我们平台的用户逐渐的增加，最先扛不住的不是我们的业务系统、不是我们自研的数据产品，而是我们的大数据集群

具体体现在多个地方

1. Yarn所有队列拥堵
   • 虽然我们Yarn已经按照任务类型、任务调度周期等特点划分了多个队列，但是在高峰期，还是会出现所有队列打满，任务积压严重的情况
   • 我们使用的调度系统是Apache Dolphinscheduler，虽然它已经足够优秀，但在某些场景下的功能依旧不够充分
     • Worker的负载均衡：针对的是Worker机器的负载，如果任务都是提交在Yarn或K8s集群上，那Worker的负载其实会比较低，这时候频繁的提交任务可能会导致Yarn、K8s的爆炸
     • Worker分组：粗粒度的分组，多个组可以共享同样的Worker，但是无法做到资源限值，可能出现别人将我的Worker资源全部占满，而我却无能为力的情况；如果我独享我的Worker，却又会出现Worker空闲的情况。总结来说就是：旱的旱死、涝的涝死
     • 不支持Hook：除非二次开发，否则某些地方想做一下特殊的事情比较困难，如果能有像Hive一样的Hook机制将会舒服很多
     • 不支持任务临时执行：可能是因为Task以Json存放在工作流相关的表中而非自己有单独的表，导致Dolphin在当前版本不支持临时执行任务，必须将任务归属到某个工作流中，通过工作流来执行
     • 缺乏Metrics信息：拿不到系统的指标信息，比如当前使用多少线程？多少任务在等待提交……等等信息；
     • 当然不是说Dolphin不好，只是他需要更多的时间去发展，但是我们公司可能等不及他的发展
   • 分析师直连Impala集群，肆意提交任务；同时又因为我们的Impala集群与Yarn机器是混布的状态，只要Impala将Io、Cpu打满，那我们Yarn上的任务必然会跑不动
2. Hdfs压力大
   • 我们Hive的默认引擎是是Hive on spark，同时我们的开发平台也支持Spark、Flink等各种类型执行引擎，很多同学在进行数据开发的时候，为了让任务跑的更快，很多情况下会将并行度开的很大；也有可能在某些场景中，需要用到多级动态分区；种种的情况，导致我们Hdfs上的小文件巨多；因为历史遗留问题，我们有很多数据都是通过Flume采集到HDFS集群，这部分数据在HDFS上产生了很多小文件
   • 我们数仓侧的H+1的ODS层任务，都是采用小时全量的方式，也就是说，每小时都会拉取一次业务库的全量数据，虽然会和前一个小时的分区进行合并，写入当天的分区，但是说到底，ODS层存储的是每天的全量业务数据，导致磁盘使用率非常高，而且有很多数据其实是冗余存在的

这些问题，我想是一家公司，从零开始做大数据都会遇到的问题，甚至说拥有成熟、经验丰富的大数据团队的公司，也会遇到这些问题。所以，为了彻底解决以上的问题，我们决定从任务、数据两个维度入手进行治理，力求做到：数据&计算资源充分利用，数据准时产出，集群平稳运行

本期先从任务治理入手，来讲讲我们是怎么进行任务治理的。

什么是任务编排

在我们看来，任务治理需要分为三步

• 任务编排
• 任务分析
• 任务全链路、全生命周期追踪

涉及到篇幅问题，本期只讲述任务编排。不过正式开始讲之前，再聊一下我们在用上任务编排系统前，存在的一些系统架构层面的问题

首先，我们能够接受任务提交并真正提交到Yarn/K8s集群上的执行器有3种

• Dolphin的Worker，负责执行Beeline(Hive)&Shell类型任务
• 基于Zeppelin封装的应用，用来提交分析型任务，如Impala、Python
• Fin-processor，我们用来提交和执行数据交换类型任务，同时也负责FLINK_SQL和SPARK_SQL类型任务的提交

可以看到，这里的设计比较混乱，能够去提交任务的系统有三个，本质上是同一件事情，却被三个应用给干了

这是其一

其二，在我们的系统Fin-processor中，每种任务类型有自己单独的线程池，当有任务被提交时，通过线程池的负载来进行任务的提交、等待、拒绝；
这样的实现方法很简单很粗暴，但是也带来了一些的问题

• 有些任务很重要，但是因为有太多的不重要的任务先提交了过来，导致重要任务被在队列中等待甚至被拒绝
• 对于即系查询任务（目前Fin上没有，但如果没有任务编排系统，未来上面也会有该类型任务），我希望很快任务就能被提交；而对于天级别任务，我的任务晚一点被提交也没问题，如何智能的去选择应该优先被提交的任务

其三，目前我们的开发平台支持配置任务的优先级和任务告警，但是因为这些配置都是人手动去配置的，每个人想怎么填就怎么填，所以导致了所有任务的优先级都是最高；而任务告警则可能告给一个毫无相关的人甚至是公司高层；这种问题我应该如何去解决？因为语言上的约束是没有用的

带着这些问题，我们再来看什么是任务编排

何为任务编排？

并非传统意义上的任务调度、容器编排，这里的任务编排主要是解决Yarn所有队列拥堵&我们系统架构设计的问题，所以我们希望我们的任务编排系统，有以下几种能力

• 细粒度资源管控：
  • 需要多种负载均衡策略，不只是执行任务的机器的负载均衡，如果任务是提交到资源调度集群如Yarn、K8s上，则需要根据对应的集群情况来判断任务是否能够提交，提交到哪一台机器……等等。
  • 支持Worker的分组，多个组之间可以共享资源，但存在优先级及最大资源限值，优先保障高优先级人群、高优先级任务被执行
  • 高优先级任务支持多种告警方式，支持值班机制；低优先级任务通过IM通知即可
• Hook机制：
  • 有别的系统需要采集我们的任务提交前、任务结束后的事件，包括一些任务信息；就像Hive hook那样，可以采集到任务血缘等信息
• 指标暴露
  • 通过指标更好的了解你系统的内部情况，提早发现问题并及时处理

这里最关键的是细粒度资源管控，因为这个任务编排系统，将会决定一个任务能否被提交，何时被提交等等

所以，接下来我们来看一下我们任务编排系统的资源组设计

资源组设计

我们的任务编排系统的资源组设计来自于Trino（原Presto），有了解过的可以跳过这部分，接下来我会说一下什么是资源组？为什么它可以帮我们做到细粒度的资源管控

何为资源组

是通过各种资源管控策略来限制任务的提交，从而降低各个集群的负载

每个人提交到任务编排系统后，会根据任务本身的属性，来给他分配到对应的资源组中。

当一个任务被提交到资源组A中时，A会检查它自身使用的资源是否已经超过了给他的限值；如果已经超过了，将会将任务放到队列中，等待资源空闲后将其取出再次提交；或者直接返回失败给提交任务的客户端

那么，资源组是通过什么指标来判断任务是否被提交呢？

资源组配置

调度策略详解

FAIR

配置为FAIR的资源组，队列中的任务按FIFO的策略来提交任务

WEIGHTED

配置为WEIGHTED的资源组，队列中的任务被提交到该资源组时，需要携带权重值，权重越高的任务约容易被提交

PRIORITY

配置为PRIORITY的资源组，队列中的任务被提交到该资源组时，需要携带优先级值，每次取出队列中优先级最高的任务提交执行

目前我们的生产环境中，一共配置了三个资源组

• tmp：临时执行的任务将被提交到该资源组中，策略为FAIR
• etl：所有的调度任务会被提交到该资源组中，策略为WEIGHTED；权重值按任务类型、任务归属BU等属性生成
• kpi：所有关键任务会被提交到该资源组中，策略为PRIORITY；通常用来保障数仓团队关键任务、老板要看的报表等；任务如果需要被分配到该组中，需要有对应的申请&审批，且如果一个任务被升级至kpi任务，那么他的所有父节点任务都会被升级为kpi任务

此外，不止资源组内部的任务有调度策略，不同的资源组之间也有调度策略；因为所有的资源组都从属于一个虚拟的root资源组，当定时任务试图取出任务提交时，会根据调度策略，来决定取出哪个资源组中的任务；因为我们想尽可能多的选择kpi资源组中的任务进行执行，所以我们对root资源组配置的策略是WEIGHTED，这样我们能保证大部分情况下，先取出kpi中的任务，而且tmp和etl中的任务也不至于太过饥饿

通过资源组，我们可以很轻易的做到任务提交管控，从而可以很好的缓解Yarn&K8s集群的压力，并且我们将所有任务的提交都收口在了我们这个系统中，我们也解决了我们架构上的一些问题，可以说，资源组是整个任务编排系统中，最核心最关键的一环

其余设计

架构设计

系统一共有四个角色，也就是说会有四个Jvm进程被启动

• ApiServer
  • 接受前端请求：任务大盘、任务执行历史查询、资源组管理、告警管理、服务管理等请求
  • 接受二方应用请求：任务的execute、kill、queryLog等
  • 请求Coordinator：负责转发任务、资源组、服务管理操作相关；因为Coordinator是高可用设计，二方应用无法得知谁是Active状态的Coordinator
  • 请求Alarm：负责转发告警相关的请求
• Coordinator
  • 接受ApiServer请求；对资源组中的任务进行CRUD
  • 接受Worker请求：主要是Worker对任务执行结果的Ack通知，任务结束后需要从Running队列中摘除该任务
  • 自身：主从切换、资源组管理、集群资源获取等；负责管理任务在提交前的所有状态
  • 请求AlarmServer：告知任务的最终状态，交由AlarmServer判断 是否需要告警、告警方式等
• Worker
  • 接受Coordinator请求：执行或杀死任务
  • 自身：提供获取当前Worker资源使用情况；负责管理任务在提交后的所有状态
  • 请求Coordinator：通过状态机的轮转，每触发一次状态的变化都会通知Coordinator
• Alarm
  • 接受ApiServer请求：告警管理；
  • 接受Coordinator请求：根据任务状态，任务告警配置，生成对应的告警策略
  • 自身：通过告警策略，将任务的结果通知给对应的责任人/责任群；

整体设计的比较灵活，如果觉得角色过多，可以考虑将Alarm和Coordinator合并

高可用&负载均衡设计

• ApiServer：所有请求先通过ELB（华为云的弹性负载均衡），再转发到ApiServer，该角色可以部署任意台
• Coordinator：通过Zookeeper的瞬时节点+监听机制可以实现主备模式；当Coordinator是备节点时，将会拒绝所有请求；当发生主从切换时，即将切换到Active状态的Coordinator，需要将处于等待提交和运行中的任务，从数据库中恢复到内存中；如果所有数据不放内存只放数据库，每次需要先扫表中所有待提交任务，再重新排序之后拿出一个任务进行提交，提交完后需要重新扫表（防止有更高优先级任务），接下来将重新排序；有多少个任务需要被提交就得重复多少次这个逻辑，时间复杂度O(n)，且影响提交性能
• Worker：上下线时，需要Kill所有由本机提交的任务，并标记为失败；
• Alarm：无状态服务，随意重启；且不在主流程中，最多无法触发告警

所有服务均会将自己的Metriccs暴露，通过Prometheus拉取，在Grafana中展示；并配置相应告警，如果服务宕机或不健康，则会通知对应的责任人（我）；

Metrics包括但不限于：Jvm状态、机器负载、磁盘状态；Coordinator服务将会上报运行中的任务数、队列中的任务数等等

任务类型

• JDBC：支持Hive/Impala/Clickhouse/Trino类型的Jdbc任务，理论上支持任意类型的Jdbc类型任务
• FLINK_SQL：支持离线&实时两种类型的FLINK_SQL任务，并支持Application/Perjob等多种模式；
• SPARK_SQL：通过PySpark封装，支持yarn-cluster模式
• 数据交换&采集：对Sqoop和Datax任务的封装
• Python：任务运行在Docker容器中，并限制使用的核数和内存数；支持任意版本的Python，支持所有Pip依赖的安装
• Shell：任务运行在Docker容器中，并限制使用的核数和内存数；支持yum install安装任意依赖包

告警设计

• 告警策略
  • 任务执行成功或任务执行失败但是是临时执行
    • 通知对应的责任人即可，且只通知一次
  • 任务非成功状态且是调度执行
    • 每15分钟触发一次告警，除非手动停止告警，否则将持续告警；停止的告警将无法恢复
    • 如果任务对应的资源组是kpi则将会打电话；否则只会发送短信

考虑到任务可能每晚都有任务会失败，有些人可能连续好几个晚上都会被叫起来处理问题，所以需要有值班体系；每晚都会有对应的值班人，任务失败优先打给值班人员，如果告警持续N次后，依旧无人关闭告警，说明值班人员未处理问题，将会打电话给兜底人，由兜底人进行处理

• 告警值班
  • 生成值班列表，每天下班前会通知对应的值班人员今天需要值班
  • 兜底人配置，当值班人员未处理任务时，由兜底人处理问题
  • 调度任务的非成功状态，将会直接发送给值班人员而非任务的owner

写在最后

• 写到这里，基本上我们任务编排系统的诞生初衷，以及相关设计，包括能解决什么样的问题，我想大家都已经明白了；那么，任务治理"三板斧"之任务编排 讲到这里就结束了，下一次我们会讲一下任务治理的下一把斧头——任务治理

• 另外，我是打算将这个任务编排系统开源的，不过肯定不能直接将公司的代码给发出来，所以我准备从零再写一份开源版本，将不会依赖公司的任何组件，预计今年结束前能够在我的Github上发出来

• 记得点赞，不要白嫖