腾讯大数据分布式任务调度平台——US

志明与数据 2023-01-19

1320

导读：今天和大家分享一下腾讯统一大数据调度平台-US，主要包括以下四大部分内容：

系统简介
系统设计
运营情况
未来规划

扫码可下载本文PPT

分享嘉宾｜马朋勃腾讯大数据高级工程师

编辑整理｜曹红姣平安人寿保险

出品社区｜DataFun

系统简介

统一调度平台是腾讯自研的一款分布式离线任务调度平台，主要负责腾讯离线分析任务的定时调度。主要起到承上启下的作用，承上即指开放接口对接数据应用平台及一些重点业务，启下即指驱动各个计算、存储资源按照任务的依赖关系有条不紊的运行。

典型场景：

该场景主要包括：数据收集、数据分析计算、数据出库 3 块内容。数据收集，支持将用户侧生成的文本数据、关系型数据、消息型数据等收集完成后，通过入库任务入库至大数据平台，即生成 ODS 层的数据。用户可以通过统一调度平台提供的计算引擎，比如 Hive、Spark 等，定义统一调度的计算任务对数据进行加工，加工后的数据即为 DWD 或 DWS 层数据。接着经过一系列挖掘后生成用户想要的有价值的数据，例如生成一张报表或游戏日活统计数据。最终通过统一调度平台的出库任务将数据推送至用户侧供其使用。目前支持关系型数据的推送或者文件系统,调用统一调度平台提供的接口进行数据的拉取。

总体而言，大数据平台就像是一个巨大的数据加工厂，原材料是用户生成的原始数据，成品就是加工提取后有价值的数据。而统一调度平台在工厂中承担一个管家的角色，协调各个组件有条不紊的运行。整个加工过程数据存在上下游依赖关系，数据依赖就直接反映出任务的依赖，基于任务的依赖关系就生成了一张复杂的 DAG 图，而统一调度平台按照调度周期驱动 DAG 图周而复始的运行。

系统设计

1. 第一代调度架构挑战

第一代调度架构设计之初能满足当时的业务需求，但随着任务体量不断增长，因为架构设计及技术选型的原因，已很难满足业务即时性的要求。主要表现为：

其一，调度核心模块扩展比较困难，单个节点承载的任务量已达到系统的极限，一

旦任务堆积可能就会引起后续任务的延时；

其二，数据库负载已达到瓶颈；

其三，没有一个完善的资源管控，因为底层资源有限，而资源消耗型的业务可能会占用大量的资源，导致时序性要求很高的任务延时；

其四，任务优先级关系管理不当，导致大量的关键任务延迟，进而造成报表延迟；

其五，系统存在大数据量处理效率不高的问题，比如任务的实例化或任务的下发等。

如下针对任务实例化及任务调度进行重点说明。针对第一代调度架构存在的问题，有两个方案可以解决此问题：其一，自研新一代调度平台；其二，采用开源调度方案。

2. 开源调度方案

业界关注得更多是调度平台的通用性，往往数据规模比较小，即未考虑海量的数据规模。例如：针对 airflow 当任务量达到百万规模时性能已达到瓶颈。另外 airflow 的维护成本非常高，因为 airflow 的文件定义是通过 Python 文件做定义，当有成百上万工作流时，就需维护成百上万的 Python 文件。另外，基于腾讯较高业务复杂度，数据体量大，集群跨地域等背景，需要设计一款支持横向扩展，且能满足腾讯集群规模和大体量任务量的任务调度平台。

3. 新一代调度系统架构

新一代调度系统重点考虑扩展性、高可用、高吞吐、灵活性 4 个指标。该系统架构引入了 BaseMaster 解决调度层的单点问题；根据业务的不同，设计不同的调度核心负责不同的业务，不同调度核心支持互相的迁移，如果任务增长即可通过扩容调度核心进行解决。对于数据库负载的问题，通过将数据库由 MySQL 更换为 Tbase（腾讯自研的分布式数据库），并且通过 Hbase 做了一个数据的冷热分离；为解决资源管控不够完善的问题，根据资源设计了基于资源的公平调度算法，限制资源消耗型业务的任务并发，避免抢占关键任务的资源造成报表延时。并且对调度内核进行了彻底的重写，提升了整体性能。

如上，为新一代调度架构系统。用户通过前台和第三方平台调统一调度平台接口层，通过网关限流、鉴权和审计之后，API 将任务定义或者工作流定义写到存储层，然后调度核心从存储层拿到对应的任务，接着调度核心再从控制层拉取对应的分析片任务，然后调度核心进行对应的任务实例化，比如：天任务每天生成一个实例、小时任务每小时生成一个实例）、依赖判断、并发控制和任务下发；第二代调度系统对比第一代调度系统做了哪些优化，具体优化措施又是如何？如下主要从实例化和任务下发两点来进行阐述。

4. 实例化

（1）实例化性能瓶颈及解决方案选型

由上图可知，任务实例生成作为调度核心的第一步动作，如果它发生了延迟，后续依赖判断及任务下发都会延迟，进而导致结果数据延迟。第一代调度系统的任务实例化存在任务实例化性能低下的问题，导致分钟级任务调度支持度低，任务延迟严重等问题。

与此同时，任务实例化存在的挑战包括：

体量大，即每天调度百万级任务，生成千万级实例，并且单库达到3亿+历史实例；
用户对实效要求高，短周期任务多，时间敏感度高；
波峰效应，高峰期瞬时实例剧增，对系统吞吐性能要求高。

通过分析发现，造成如上痛点的主要原因包括：

应用负载高：调度内核负载高且水平扩展困难，百万任务单机调度；
存储负载高：数据库单机已达到瓶颈，虽然做了读写分离，但是单节点还是需要保存所有数据；
程序性能差：由于生成和写入数据库是串行操作，导致大数据事务，而且回滚又会造成大批量任务的延时。

基于上述问题，我们探讨了业界常用的解决方案：

① 存储可伸缩

非代理模式的分库分表：应用和数据是放在同一个进程，它的优点是分片算法非常丰富，业务可以自定义分片逻辑；由于没有多余的服务进程，它的运维成本比较低，但是分片逻辑和业务耦合度高，导致泛用性较差。
代理模式的分库分表：DAL（数据接入层）作为单独的一个进程存在，对业务的侵入低，但是需要维护 DAL 服务，运维成本相对而言高一些。
分布式方案：对业务完全无感知，由 DB 层维护分布式事务和分片相应的逻辑，对硬件成本要求比较高。

考虑到业务的侵入性和扩展性，最终选择了分布式方案，即采用 Tbase 做存储。Tbase 是基于 PG 自研的一款分布式数据库，由专业的运维人员去维护。

② 应用可伸缩

水平伸缩：通过增加机器的方式来分担系统的压力，对架构设计要求相对较高。
垂直伸缩：通过提高单机的性能，包括对软件的优化及硬件的升级，由于主要从软件层面，因此对开发人员的要求比较高。

基于腾讯体量和成本的考虑，应用可伸缩这块在水平伸缩和垂直伸缩都有落地实现。接下来我们以实例化为例，看下具体的优化措施。

（2）实例化——解决方案

就存储可伸缩做的具体优化包括以下 3 点：

① 为解决 MySQL 单点性能的问题，将 MySQL 切换为分布式 Tbase，支持数据库的水平扩展；

② 为解决查询效率的问题，对数据做了冷热分离，冷数据采用 Hbase 做存储；

③ 为解决水平扩展，引入了 BaseMaster，即每个调度核心在启动之后从 BaseMaster去获取对应的任务分片，任务调度核心只负责对应分片的任务。

应用垂直伸缩方案即是针对任务实例生成进行优化，生成实例的过程优化包括对实例进行 Hash 分桶。Hash 算法可以自定义，并且每个桶可以自己定义桶的大小和缓冲时间，当桶满了或者达到它的缓冲时间之后，对这个桶做一个 mini-batch 然后做批量提交，因为是有多个桶，就将之前的串行一次写入就转换为 mini-batch 的并行写入，进而实例化性能得到大幅提升。

5. 任务下发

（1）任务下发——问题分析

任务下发主要存在报表延时和服务压力大两个问题。具体表现为：① 偶发性重点报表达到分钟级左右的延时；② 服务压力大。

任务下发面临的挑战包括：① 任务量大，达到千万级的任务下发；② 用户对数据实效性要求较高；底层任务依赖较繁杂，需要依赖计算、存储、权限系统及用户侧的OLAP等；③ 底层资源有限；④ 任务依赖繁杂，在有限的时间及资源内驱动繁杂的组件运行上千万的任务。

分析报表延的案例发现造成报表延时的原因包括：① 未考虑链路依赖的优先级；比如：低优先级的父任务导致高优先级的子任务不能及时运行；② 低优先级的任务长时间得不到运行导致饿死的情况；③ 用户对于小周期任务延时更加敏感；例如：因为调度时将月/天/小时任务一起执行，导致用户对小周期任务的延迟更加敏感；④ 任务的依赖链路太长，导致下游任务不能及时运行，从而导致报表延时。

分析服务压力大的案例发现主要原因是没有对服务器做并发控制，大量的任务可能会压垮服务器。例如：某个出库任务，将计算之后的数据出库到 MySQL，这时大量的任务同时对多个表执行出库任务，这时可能会把用户侧的 MySQL 服务压死，这时需要限制服务的并发。

而上一代调度算法采用优先级调度算法，进一步来说，优先级采用的是静态优先级。如下图所示，A 任务的优先级低，B 任务的优先级高，但是 A 任务下游有两个高优先级的任务，而 B 任务的下游没有高优先级任务，按照上一代调度算法，B 要优于 A 运行，而实际理想的情况是，A要优于 B 运行。

上一代调度算法支持任务并发控制，不支持服务并发。例如如下图所示：在上一代调度算法中，X 下发两个正在运行的实例后，然后 Y 还可以下发两个任务去运行。理想的情况是，需要等到 X 的所有任务执行结束之后，然后 Y 才能下发。

所以，最终得出的结论是问题出现在任务调度算法上。

（2）任务下发——方案选择

由上可知，任务下发的调度算法主要从两个方向进行优化：① 动态优先级，优先级考虑任务动态执行过程；② 动态并发控制，根据资源实时动态做并发控制。

参考操作系统的任务调度方案：① FCFS（First Come First Service），该方案优点是公平，但是该方案对短作业不友好；② SJF（Short Job First），该方案优点是全局等待时间最短，缺点是对长作业不友好；基于第①和第②方案都不能满足统一调度需求，只能做到局部最优。参考前两个方案设计了基于资源的公平调度算法即第③个方案。③ RFS（Resource Fair Schedule），综合考虑动态优先级（基于任务执行过程动态计算优先级）和资源的并发控制（执行过程动态执行动态并发）。

为实现第③个方案，带来了两个问题，如何动态计算优先级？什么是资源，如何做资源的并发控制？

a. 任务下发——动态优先级

实现任务下发动态优先级的计算原则包括：① 任务的紧迫性，任务越接近Deadline，它的优先级越高；② 任务的关键性，即任务所对应的业务越关键则任务越关键，优先级越高；③ 任务的频繁性，即任务越频繁优先级越高，即分钟任务的优先级>天任务的优先级>周任务的优先级>月任务的优先级；④ 周期的快捷性，短作业任务优先级>长作业任务优先级；⑤ 依赖的传递性，越是上游的任务优先级越高，因为下游的任务都依赖上游的任务。

基于上述原则，整理出任务优先级要素：

延迟时间（紧迫性）：延迟越久，优先级越高
静态优先级（关键性）:业务关键任务
周期（频繁性）：分钟>小时>天>周>月
任务类型（快捷性）：check（检查）>sync（同步）>calculating（计算）
链路（传递性）：下游任务链路深度

b. 任务下发——动态资源控制

面临的主要挑战是，底层服务类型非常多，方案很难统一。比如任务下发到一个 Executor（执行机，即包括物理机和容器）上，那 Executor 需要做资源并发控制，不能下发太多的任务把Executor资源压垮；同理，针对 Yarn 的资源队列，比如提交任务到 Yarn 上去，提交太多任务那就会造成任务阻塞，这时需要控制 Yarn 资源队列（Queue）的并发；包括 Hive 集群、HDFS 集群或者用户侧的 MySQL 服务（Service）等，都需要做资源并发控制。另外，随着服务的资源消耗，需要动态计算任务的资源情况，它的实时性要求比较高。这块的解决方案是采用领域模型对服务进行一个高度抽象，即一切皆是资源（参考 Linux 操作系统一切皆是文件的思想），抽象一个资源接口，包括已用配额、最小配额和最大配额，根据服务资源量化为配额，即资源配额。于此同时，根据执行历史任务评估任务占用配额。

（3）任务下发解决方案——资源公平调度

确定任务动态优先级和动态资源控制，我们看下基于资源的公平调度算法具体实现；资源公平调度的核心思想是如何从百万级任务中获取满足资源并发控制的最高动态优先级任务。

最高动态优先级任务分为如下 5 个步骤：

① 分片加载：调度内核加载对应分片的待执行任务；

② 分桶：待执行任务 Hash 分桶；

③ 桶内排序：每个桶按照优先级进行桶内排序；

④ 桶内并发控制：桶内按照动态优先级从高到低遍历，判断资源并发控制，直到有一个任务满足并发控制；

⑤ 桶外排序：对每个桶满足并发控制的高优先级任务排序获取优先级最高任务。

运营情况

1. 实例化落地效果

① 性能提升：性能提升 30 倍；百万任务实例化性能提升了 10 倍+

② 支持分钟级调度：满足业务分钟级调度需求

2. 基于资源公平调度——落地效果

① 用户重点报表均按时输出，满足用户分钟需求

② 业务侧未出现因任务并发大引起的宕机现象

3. 运营现状——内网

① 核心指标：千万级日调度任务、万级用户规模、上千的集群规模、支持80+任务类型

② 每天调度任务数：千万级，任务数年增长50%

③ 服务部门/业务：IEG、PCG、CSIG、WXG、CDG等

4. 赋能公有云

研发 WeData 数据开发平台，目前已开放使用，它是一款一站式的数据协作开发平台，包括数据分析、工作流协同编排、数据资产管理、数据治理等全链路的数据加工能力，帮助数据工程师高效构建企业级的数据中台。

项目未来规划

短期规划：持续系统打磨，极致提升性能及用户体验；加强系统自运维，提升问题诊断分析能力；

长期规划：打造一站式数据开发平台。

问答环节

Q1：调度的周期拉取是轮询模式还是 Trigger 模式，系统语言是采用哪种语言编写？

A1：采用 Java 语言编写。主要是支持两种模式，轮询模式主要针对大批量长周期的，对实时性不太敏感的任务调度采用轮询模式，对于短周期分钟级别实时性要求高的任务调度采用触发（加载系统内存的轮询，对系统内存消耗大）的模式。

Q2：海量任务数据存储是如何解决的？

A2：百万任务每天会生成千万级别的实例，存储做了冷热分离，把最近两天且状态处在待运行的任务或者依赖判断之前的任务全部存在 Tbase（腾讯自研的分布式数据库）中。将已经运行成功的或者永久终止（重试很多次已失败）的任务存储在 Hbase 中。

Q3：调度平台怎么实现自动扩容的？

A3：腾讯调度平台引入了 BaseMaster，BaseMaster 的节点是支持热备的，每个调度核心从 BaseMaster 去拉取对应任务的分片，不同调度分片的任务分片如果宕机是可以水平迁移的。当扩容一个节点进来时，只需从 BaseMaster 去拉取对应的分片，然后 BaseMaster 就会从一个节点获取部分任务分配到新节点上去。

Q4：任务的中间输出数据（二进制文件）是如何解决的？

A4：针对各种组件，如Hive中间输出数据会存储在中间临时表，任务执行完成后会把临时数据表 drop 掉；另外，Spark 这种会有一个临时的 HDFS 目录，用户把数据写到临时的 HDFS目录，操作完成后会把临时目录进行清理；另外在计算过程中，为保证数据一致性，对 Hive 等组件也做了一些改造，执行完成后，会把数据会做一次性的 move 操作。

Q5：版本更新时如何解决已运行但未执行完成的作业任务调度问题？

A5：1、调度核心如果重启，如果这时作业状态已完成就会去提交状态，这时会不停重试提交状态，重试的步长也会不断放大，当调度起来后，就会把状态提交上来。2、如果 DB 故障，BaseMaster 写 DB 失败后会将数据写入磁盘，待 DB 恢复之后，会从磁盘读取数据然后重新写入 DB；3、不同的调度核心可以进行水平迁移，当任务 A 的调度核心停止后，会把任务迁移到其它的调度核心，只要有一定量的调度核心是存活，就能保证任务的正常调度。

今天的分享就到这里，谢谢大家。‍‍‍