Druid 索引服务的资源精细化调度

本文将从三个方面讨论关于Druid 索引服务的资源精细化调度：

一、Druid索引服务的简介

介绍了Druid 索引服务的架构和调度细节。（介绍原理）

二、实践中出现的问题

经过线上生产环境的实践，旧的调度方案出现的问题。（提出问题）

三、资源精细化调度的设计与实现

提出了资源精细化调度方案的设计与实现对问题进行了解决，并做了实际效果评估。（解决问题）

一、Druid索引服务的简介

索引服务是Druid提供用于导入并创建segments 数据文件的服务，索引服务是一个高可用的分布式服务，并以主从结构作为架构模式，索引服务由三大组件构成，overlord作为主节点，middlemanager是从节点，peon用于运行一个task。

1、索引服务三大组件

（1）overlord : 负责创建task、分发task到middlemanager上运行，为task创建锁以及跟踪task运行状态并反馈给用户。

（2）middlemanager : 作为从节点，负责接收主节点分配的任务，然后为每个task启动一个独立的JVM进程来完成具体的任务。

（3）speon : 由middlemanager启动的一个进程用于运行一个task任务。

其中task的类型有很多，其中包括：

index hadoop task : 是Hadoop的索引任务，利用Hadoop集群执行MapReduce任务以完成segment数据文件的创建，适合体量比较大的segments数据文件的创建任务，一般我们的公司的离线任务基本上都是周期性调度index_hadoop task来完成数据的导入。

index kafka task : 专门用于Kafka数据的实时摄入，通过Kafka索引服务可以在Overlord上配置一个KafkaSupervisor，通过管理Kafka索引任务的创建和生命周期来促进Kafka的摄取。这些索引任务使用Kafka自己的分区和偏移机制读取事件，因此能够提供完全一次摄取的保证。他们还能够从Kafka读取非近期事件，并且不受其他摄取机制强加的窗口期限的影响。

merge task : 是一个合并索引任务，将多个segments数据文件按照指定的聚合方法合并为一个segments数据文件。

kill task : 销毁索引任务，将执行时间范围内的数据从Druid集群的深度存储中删除。

2、overlord基于槽位的调度过程

如图所示：

每台middlemager设置固定的capacity（能调度的最大任务个数），overlord基于一定策略（如剩余可用capacity），分配任务交由middlemanager执行。

1）备选worker的过滤

版本过滤：当一个middlemanager 在灰度升级的时候一般会先发送disable命令给middlemanager，middlemanager会把自己的版本标记为空字符串，overlord在调度的时候过滤该版本的worker。

黑名单过滤：overlord将对失败次数达到一定阀值的middlemanager列为黑名单，默认黑名单的比例不会超过20%，并且被列为黑名单的middlemanager会被定期回放到白名单

槽位空闲过滤：如果该middlemanager的没有空闲槽位来满足运行一个task的时候就会被过滤掉。

2）worker的选择策略：

最大适应算法: 选择可用的空闲槽位最多的middlemanager，这样能够使得分配更加均衡。
最佳适应算法: 选择空闲槽位最小并且能够运行task的middlemanager，这适用于开启了弹性自动伸缩的情况，尽可能留出一些空闲的middlemanager使得暂时退出。
自定义算法：自己实现的JavaScript用于woker策略的选择

3）现有内存调度方案

当middlemanager接受到一个task的时候，其槽位数将会减1，middlemanager 进程会以命令行的方式

io.druid.cli.Main internal peon <task_file> <status_file>

创建一个新的peon进程，在执行命令中会分配该进程的内存大小，主要参考以下配置：

（1）middleManager.runtime.properties {druid.indexer.runner.javaOpts} 中的 -Xmx5g（堆内）和

-XX:MaxDirectMemorySize=5g（堆外）

（2）task payload 里面的context{druid.indexer.runner.javaOpts}参数

我们的采用集群的全局配置，所有任务均按10G的量分配资源，堆外为5GB，堆内为5GB。

二、实践中出现的问题

由于我们集群的全局配置相同，所有任务均按10G分配资源，这种笼统的资源分配方式造成了一些问题：

1）严重的资源浪费

实际上hadoop index任务在运行过程并不需要太多内存，并且不同流量的kafka任务实际消耗内存差异较大。把所有的task的内存的都统一设定为10GB会造成极大的内存浪费。

2）节点内存实际使用无法预估

因为middlemanger只是指定了固定的槽位数，具体task内存分配只是按照上面两种方式，如果统一为10GB那么middlemanger的内存上限是: 槽位数* 10GB，但是因为task的调度是随机的，并且用户可以自己在context指定其大小，所以middlemanger内存真实使用情况将会不可预估, 甚至可能还会造成进程的OOM。

三、资源精细化调度的设计与实现

1、资源的精细化分配的意义

随着业务需求日益旺盛，数据规模不断膨胀，task的数据越来越多，假设一台middlemanger进程最多被分配150GB内存，槽位最多为15个，我们在同一个时刻需要跑2000个task，那么需要至少133台节点同时运行，资源的消耗是比较大的，线上偶尔发生的middlemanger OOM导致task失败，造成kakfa 实时任务的延迟也是血的教训。

对不同类型的task进行按需申请，有计划地分配资源对整个集群的稳定性、可扩展性、和资源高效利用具有重大的意义。

我们对线上的实时和离线的task 进行了统计，发现我们确实造成了非常巨大的资源浪费：

如图：

上图的task分配量是根据集群的middlemanger的JVM的配置参数确定，使用量是通过统计了近7天的task的内存使用量的峰值，可以看出实时和离线的内存的实际分配和实际使用之前差距有点大，使用率并不太高。

上面表格数据显示，实时task的内存使用率在22.16%，离线的task的内存使用率在0.87%，确实造成巨大的资源浪费。

2、资源的精细化分配方案

新旧方案的对比如下：

每台middlemanager配置的capacity，不再指槽位，而是指内存（单位G )

以前的配置：（槽位为8个）

ex: druid.worker.capacity=8，槽位数本应该根据middmanager实际被分配内存大小做的设置，但是因为一个节点上可能运行多个角色，所以可能存在超配。

之后的配置：（内存为80G）

ex: druid.worker.capacity=80

我们需要修改overlord和middlemanager的代码，使得overlord基于worker的Capacity，以及task的requiredCapacity进行task分配，每个task的requiredCapacity是指需要的内存G数，并且在overlord的控制台页面显示的capacity, currCapacityUsed 的含义将发生变化，之前是指总槽位数和剩余槽位数，现在分别指当前middlemanager内存总量(GB)，已经使用量(GB), 当version为1时表示资源精细化调度的版本，当version为0表示槽位调度的版本，如下图所示：

我们设计的task的requiredCapacity 有几种配置方式如下图：（优先级低的可以被优先级高的覆盖）

（1）默认配置 (基于task type)（优先级低）

a. kafka task:

onheap.memGBs : 5

offheap.memGBs : 5

b. hadoop task:

onheap.memGBs : 1

offheap.memGBs : 1

c. other task:

onheap.memGBs : 3

offheap.memGBs : 3

（2）套餐分配: 提交index task的spec配置文件的套餐分配 : (优先级中）

context中配置：druid.indexer.fork.property.druid.task.mem.level : 1 // 取值分别代表低，中，高

默认的套餐：

level = 1 低 : -Xmx1g -XX:MaxDirectMemorySize=1g 堆外1G+ 堆内1G = 2G

level = 2 中 : -Xmx3g -XX:MaxDirectMemorySize=3g 堆外3G+ 堆内3G = 6G

level = 3 高 : -Xmx5g -XX:MaxDirectMemorySize=5g 堆外5G+ 堆内5G = 10G

（3）精确配置: index task的配置文件的独立精确配置:（优先级高）

每个任务支持独立设置堆内与堆外内存（单位G，整数，可在context字段增加如下属性)

spec.context中配置：

druid.indexer.fork.property.druid.task.onheap.memGBs : 5, → -Xmx5g

druid.indexer.fork.property.druid.task.offheap.memGBs : 5, → -XX:MaxDirectMemorySize=5g

参数的可配置性 , 在_common中配置的内容如下：

conf/druid/_common/common.runtime.properties

基于不同level的Task的值：

"druid.indexer.task.taskOnHeapGBsByLevel " : {"1":1, "2":3, "3":5}

"druid.indexer.task.taskOffHeapGBsByLevel" : {"1":1, "2":3, "3":5}

基于不同类型的Task的默认值：

"druid.indexer.task.taskDefaultOnHeapGBsByType" : {"index_kafka" : 5, "index_hadoop":1, "_common_task":3}

"druid.indexer.task.taskDefaultOffHeapGBsByType" : {"index_kafka" : 5, "index_hadoop":1, "_common_task":3}

heap的最大值的配置

"druid.indexer.task.taskMaxHeapGBs" : 10

3、兼容性的实现

因为线上环境的升级无法全停全起，需要滚动升级，需要对overlord和middlemanager做兼容性的设计。

我们的灰度升级方案是：

第一步：先升级overlord

overlord会对新提交的task进行资源的精细化调度的逻辑，根据集群汇报上来的middlemanager 和已经被分配的task的数量做调度参考。由于此刻middlemanager是旧版本的节点，需要保证新分配的task能够在旧集群上运行，在web页面展示的时候能够灵活反映middlemanager的实际内存使用而不是槽位数。

第二步：灰度升级middlemanager

部分middlemanager的升级导致上报到overlord的middlemanager节点的版本多样性。所以需要保证新旧版本的middleManager具有版本的标识，通过设置version的值就可以达到。

middlemanager节点有个配置项：druid.indexer.task.restoreTasksOnRestart=true，如果为true，则middleManagers将在stop的时候会把当前的task缓存起来，并在重新启动时恢复它们。这样就造成了一种现象，新版的middleManagers上有不同版本的task，我们需要对新版的middleManagers或者新版的overlord端能够识别是否是旧版本的task并且在上报task的时候准确传达内存使用信息。

新旧版本上的middleManager上都可能存在新旧版本的task的混合运行状态，但是调度到旧版本上运行的task都是按照之前的逻辑进行。

兼容性要点：

调度的时候： 1、overlord具有混合middlemanager的调度能力

一个task被分配到不用版本的middlemanager所请求的capacity的含义 : 槽位数和内存数

上报的时候： 2、middlemanager具备混合task汇报能力

3、overlord 具备混合版本的middlemanager 上报的辨别。

上图显示的正在灰度升级的中的集群状态，如果一个新ovelord上分发一个task到旧版本的middlemanger上，task的requereCapcacity=10，但是旧版本的middlemanger在启动进程的时候不会参考这个值，并且上报到zookeeper上的requiredCapacity为10，而之前运行的旧的task的requiredCapacity=1， overlord是如何辨别这种差距的呢？我们这边的优化是代码如下：

判断如果是旧版本的task，则忽略requiredCapacity参数，直接计算task的总数再乘以10GB，作为该middlemanger的当前内存使用量。
如果一个新ovelord上分发一个task到新版本的middlemanger上，task的requereCapcacity=8，新版本的middlemanger在启动进程的时候会参考这个值，并且上报到zookeeper上的requiredCapacity为8，而之前运行的旧的task的requiredCapacity=1，新版本的middlemanger会按照旧版本的形式去调度，为10GB。而overlord是如何辨别这种差距的呢？我们这边的优化也是如上, 如果为1,就确定为10GB。

ovelord在调度的时候会检查每一个middlemanger的内存使用量，并且计算task的 requiredCapacity，针对这个task是否是调度在旧版本middlemanger做了区分：

4、上线实际收益评估

通过上面的设计方案对Druid代码进行改进，对实时的task采用最近7天的峰值进行精确化分配，离线的task的内存分配采用了套餐的分配模式，统计结果发现，收益是显著的，实时任务的内存节省率达到42%，离线的task能够节省80%的内存。

下面我们选出的部分的实时和离线的数据源做统计：

假设一台middlemanger进程最多被分配150GB内存，之前最多只能跑15个实时或者离线的task，现在一台同样的middlemanger可以跑26个实时任务 75个离线任务，我们在同一个时刻需要跑2000个task，其中（实时的1200, 离线800）之前至少133台节点同时运行，现在我们最多需要58台机器。节省了75台机器。

Druid 索引服务的资源精细化调度相关推荐

Druid Kafka索引服务的Task动态伸缩
一.Kafka Indexing Service 运行原理 1.简介 Kafka Indexing Service 是 Druid 推出的利用 Druid 的索引服务实时消费 Kafka 数据的插件. ...
如何通过 Serverless 技术降低微服务应用资源成本？
前言在大型分布式 IT 架构领域,微服务是一项必不可少的技术.从本质上来讲,微服务是一种架构风格,将一个大型的系统拆分为多个拥有独立生命周期的应用,应用之间采用轻量级的通信机制进行通信.这些应用都是 ...
通过Serverless技术降低微服务应用资源成本
简介:从本质上来讲,微服务是一种架构风格,将一个大型的系统拆分为多个拥有独立生命周期的应用,应用之间采用轻量级的通信机制进行通信.这些应用都是围绕具体业务进行构建,可以独立部署.独立迭代,也可能根据业 ...
过Serverless技术降低微服务应用资源成本
前言在大型分布式IT架构领域,微服务是一项必不可少的技术.从本质上来讲,微服务是一种架构风格,将一个大型的系统拆分为多个拥有独立生命周期的应用,应用之间采用轻量级的通信机制进行通信.这些应用都是围绕 ...
华为云服务权限在哪_华为云数据湖探索服务DLI，精细化保障企业大数据安全
原标题:华为云数据湖探索服务DLI,精细化保障企业大数据安全随着企业业务的不断发展,企业大数据资产在企业辅助决策.用户画像.推荐系统等诸多业务流程中扮演着越来越重要的作用,如何保证企业大数据在满足各 ...
论文：Elastic Scheduling for Microservice Applications in Clouds （云环境下微服务应用的弹性调度）
Elastic Scheduling for Microservice Applications in Clouds (云环境下微服务应用的弹性调度) 摘要: 微服务被广泛用于灵活的软件开发.最近,容 ...
如何制定一个有效的项目管理计划？目标可预期、资源可调度、变化可控制、问题可预见、业绩可评价
有人将很多项目的管理方式归结为"六拍运动"我深有同感,特将其整理如下,希望我们能够警惕这种 "运动". 第一拍:拍脑门.经常有些领导有了做一个项目的想法后,不 ...
干货 | 跨多业务线挑战下，携程订单索引服务的1.0到2.0
作者简介唐巍,携程用户平台部订单服务组资深后端开发,在互联网尤其是移动互联网方面有丰富的经验,目前主要负责OrderIndex的维护和架构升级工作. 经过团队几个月的努力,我们最近终于完成了OI(订 ...
win7关闭系统索引服务器,如何优化Win7系统之如何关闭索引服务
Win7如何优化系统? 怎样加快Win7系统优化? Win7系统优化怎么做? 能够使用Windows7操作系统成为了许多电脑用户的一大喜悦之事,相比之前的XP和Vista系统,Windows7系统真的 ...

Druid 索引服务的资源精细化调度

Druid 索引服务的资源精细化调度相关推荐

最新文章

热门文章