ResourceManager中的Resource Estimator框架介绍与算法剖析

原创

宋超

修改于 2018-08-12 10:24:25

3.4K8

修改于 2018-08-12 10:24:25

文章被收录于专栏：公有云大数据平台弹性 MapReduce公有云大数据平台弹性 MapReduce

本文首先介绍了Hadoop中的ResourceManager中的estimator service的框架与运行流程，然后对其中用到的资源估算算法进行了原理剖析。

一. Resource Estimator Service的出发点与目标

估计作业运行使用资源是大数据处理集群的一个重要且具有挑战性的问题。随着用户使用的集群资源越来越多，这一需求被逐渐放大。当前现有的解决方案一般是依赖于用户的经验来对作业资源需求进行估计，这样即繁琐又低效。根据对集群工作负载的分析，可以发现大部分工作(超过60%)是重复工作，这样我们便有机会根据作业历史资源使用情况来估计作业下一次的资源需求量。同时，在未来，希望能提出一种与框架无关的黑盒解决方案。这样，即使作业来自不同的计算框架，我们也能对重复性作业进行资源需求估算。

二. Resource Estimator Service的框架结构

Hadoop-resource estimator主要由三个模块组成：Translator，SkylineStore和Estimator。下面分别介绍这三部分。

1.ResourceSkyline用来表征作业在其生命周期中的资源利用率。它使用RLESparseResourceAllocation记录容器分配的信息。RecurrenceId用于标识重复pipeline的特定运行。pipeline可以包含多个作业，每个作业都有一个ResourceSkyline来表征其资源利用率。

2.Translator用来解析作业日志，提取他们的ResourceSkylines并将它们存储到SkylineStore。SingleLineParser解析日志流中的一行并提取ResourceSkyline。

3.SkylineStore充当Hadoop-resource estimator的存储层，由2部分组成。HistorySkylineStore存储由转换程序提取的ResourceSkylines。它支持四种操作：addHistory，deleteHistory，updateHistory和getHistory。addHistory将新的ResourceSkylines附加到定期pipeline，而updateHistory删除特定定期pipeline的所有ResourceSkyline，并重新插入新的ResourceSkylines。PredictionSkylineStore存储由Estimator生成的预测RLESparseResourceAllocation。它支持两个操作：addEstimation和getEstimation。

4.Estimator根据历史记录运行预测重复出现的pipeline资源需求，将预测存储到SkylineStore并在YARN上进行资源预留。Solver读取特定定期pipeline的所有历史ResourceSkylines，并预测其包含在RLESparseResourceAllocation中的新资源需求。目前，Hadoop-resource estimator提供了一个LPSOLVER来进行预测（其中用到的算法模型会在后面进行讲解）。

三.以示例demo演示其运行流程

Resource Estimator Service的URI是http://0.0.0.0，默认服务端口是9998

(在$ ResourceEstimatorServiceHome/conf/resourceestimator-config.xml” 中配置)。在$ ResourceEstimatorServiceHome/data中，有一个示例日志文件resourceEstimatorService.txt，其中包含2次运行的tpch_q12查询作业的日志。进行资源预测主要有以下几个步骤：

1.解析作业日志：

POST http://URI:port/resourceestimator/translator/LOG_FILE_DIRECTORY

发送

POST http://0.0.0.0:9998/resourceestimator/translator/data/resourceEstimatorService.txt

underlying estimator将从日志文件中提取ResourceSkylines并将它们存储在jobHistory SkylineStore中。

2.查询作业的历史ResourceSkylines：

GET http://URI:port/resourceestimator/skylinestore/history/{pipelineId}/{runId}

发送

GET http://0.0.0.0:9998/resourceestimator/skylinestore/history/*/*

underlying estimator将返回历史SkylineStore中的所有记录。在示例文件中能够看到两次运行tpch_q12的ResourceSkylines：tpch_q12_0和tpch_q12_1。

3.预测作业的资源使用情况：

GET http://URI:port/resourceestimator/estimator/{pipelineId}

发送

http://0.0.0.0:9998/resourceestimator/estimator/tpch_q12

estimator将根据其历史ResourceSkylines预测新运行的作业资源需求，并将预测的资源需求存储到jobEstimation SkylineStore。

4.查询作业的预测资源情况：

GET http://URI:port/resourceestimator/skylinestore/estimate/{pipelineId}

发送

http://0.0.0.0:9998/resourceestimator/skylinestore/estimation/tpch_q12

估算器将返回tpch_q12作业资源预测情况。

5.删除作业的历史资源情况数据：

DELETE http://URI:port/resourceestimator/skylinestore/history/{pipelineId}/{runId}

发送

http://0.0.0.0:9998/resourceestimator/skylinestore/history/tpch_q12/tpch_q12_0

underlying estimator将删除tpch_q12_0的ResourceSkyline记录。重新发送

GET http://0.0.0.0:9998/resourceestimator/skylinestore/history/*/*

underlying estimator只返回tpch_q12_1的ResourceSkyline。

四.资源预测算法中用到的数据介绍

Hadoop-resource estimator的Translator组件会解析日志并将其按照一定规范的格式进行拼接，下面给出了示例中的资源历史使用数据和预测资源数据，可以看到作业的历史资源使用数据是同一个job的两次run，分别为tpch_q12_0和tpch_q12_1,其主要给出了随时间变化的memory和cpu的使用情况。其中第0时间单位表示的是container规格，为memory:1024,vcores:1，第25时间单位为作业结束时刻，memory和cpu皆为0。可以看到预测数据根据历史数据给出了10~25时间单位的资源预测数据。

历史资源使用数据:

[[{"pipelineId":"tpch\_q12","runId":"tpch\_q12\_0"},

[{"jobId":"tpch\_q12\_0","jobInputDataSize":0.0,"jobSubmissionTime":0,"jobFinishTime":25,"containerSpec":{"memory":1024,"vcores":1},

"skylineList":

{"resourceAllocation":{

"0":{"memory":1024,"vcores":1},

"10":{"memory":1099776,"vcores":1074},

"15":{"memory":2598912,"vcores":2538},

"20":{"memory":2527232,"vcores":2468},

"25":{"memory":0,"vcores":0}}}}]],

[{"pipelineId":"tpch\_q12","runId":"tpch\_q12\_1"},

[{"jobId":"tpch\_q12\_1","jobInputDataSize":0.0,"jobSubmissionTime":0,"jobFinishTime":25,"containerSpec":{"memory":1024,"vcores":1},

"skylineList":

{"resourceAllocation":{

"0":{"memory":1024,"vcores":1},

"10":{"memory":813056,"vcores":794},

"15":{"memory":2577408,"vcores":2517},

"20":{"memory":2543616,"vcores":2484},

"25":{"memory":0,"vcores":0}}}}]]]

预测数据:

{"resourceAllocation":

"10":{"memory":1083392,"vcores":1058},

"15":{"memory":2598912,"vcores":2538},

"20":{"memory":2543616,"vcores":2484},

"25":{"memory":0,"vcores":0}}}

五.Resource Estimator Service算法框架与原理

在本部分将重点介绍一下estimator中用到的资源预测算法原理。此算法由微软提出，其链接在文末参考资料中给出。下图是estimator的运行框架，可以看到其主要由三部分组成，下面分别介绍这三部分。

Automatic interence,提取出作业的运行时间和历史资源使用情况。 (a) Extractor of target,能提取出作业的运行开始与结束时间。 (b) Job resource model,能提取出作业的资源使用情况,例如作业资源随时间运行的变化情况和资源使用总量。
Recurring Reservation,此部分包括有Job Resource Model，可以根据作业历史运行时间与作业历史资源使用情况给出下一任务的资源使用情况。 (a) 通过改变参数α，可以控制estimator在分配资源的时候是侧重过分配还是侧重欠分配。 (b) 根据作业资源预测模型给出的预测值为作业在原来分配的资源的基础上添加资源添加agenda。此job下一个run就运行在此资源分配的基础上。
Dynamic Reprovisioning，此部分根据前面给出的资源agenda,动态调整作业的每个运行阶段的资源分配。

六.算法原理剖析

微软提出的此资源分配算法本质上是一种最优化算法，其优化的目标函数是由两部分组成的线性组合，下文中stage的概念是指每个job的运行期间按照一定规则划分成多个时间片，每个时间片称之为一个stage，下面分步骤阐述其算法原理。

1.首先定义一个目标函数，也可以称之为损失函数，即我们优化的目标。在此算法中由过分配和欠分配组成的线性组合组成损失函数costfunction。目标就是minimize(cost = αA_{0}(s)+(1-α)A_{u}(s))。其中A_{0}(s)表示在当前stage的资源过分配值，其是由当前stage的分配值减去此stage的历史资源使用均值然后取平均得到，其公式表示为A_{0}(s)=\frac{1}{N}\sum_{i=1}^N\sum_{k}(s_{k}-s_{i,k})^{+},s_{k}即为当前的资源分配值，s_{i,k}即为第i次run的历史资源使用值;A_{u}(s)表示当前stage的欠分配值，其是由上一stage的欠分配值加上当前stage的欠分配值得到，公式表示如下：D_{i,k}(s_{1},...,s_{k})=(D_{i,k}+s_{i,k}-s_{k})^{+},A_{u}(s)=\frac{1}{N}\sum_{i=1}^{N}D_{i,k}(s),下图比较直观的显示了estimator在预测资源时的一种过分配与欠分配的情况。

2.针对每个stage,此算法的策略就是选择可以使得costfunction最小的资源分配方式，即选择一个值使得costfunction最小，即得到S_k,即每一个stage上的资源分配值。因为分配值是固定规格的倍数，所以在实现时可以通过简单的for循环或者一些最优化算法比如爬山法或者蚁群算法就可以快速得到最小值。

3.总结：算法中的做法是针对一个job，根据其历史运行时间拿到其作业开始和结束时间，在这时间段内按照一定规则划分时间片，每一个时间片为一个stage，根据同一job多次run的历史资源使用情况来预测下一run的资源使用情况。其每次配置的策略是使得costfunction最小。costfunction是过分配与欠分配的一个线性组合。

七.算法的测试效果

在本次测试中运行tpch_q12作业9次，并在每次运行中收集作业的资源skylines。然后，在Resource Estimator Service中运行日志解析器，从日志中提取ResourceSkylines并将它们存储在SkylineStore中。下面绘制了作业的ResourceSkylines以进行演示。