一种基于能耗感知的Spark节能调度方法

摘要

本发明公开了一种基于能耗感知的Spark节能调度方法。首先构建Spark计算框架下大数据计算能耗模型，基于该模型建立任务与计算资源的能耗和执行时间关系策略表，通过策略表指导并优化Spark任务调度，在保证并行计算效率的前提下有效降低计算总能耗。本发明解决了Spark原有调度策略无法感知能耗的缺陷，该方法具有能耗感知，动态优化调度和高可扩展的特点，有效降低运行在Spark计算框架下的应用程序产生的能耗。

说明书

技术领域

本发明涉及大数据处理领域和能效领域，具体是一种基于能耗感知的大数据能耗模型和基于该模型的Spark计算框架下的节能调度策略。

背景技术

大数据计算产生巨大电能消耗已经成为数据中心亟待解决的问题。目前许多企业和组织机构都面临大规模数据计算的问题，在考虑计算效率的同时，计算成本也是其关心的重要方面。企业和组织都希望能够降低大数据计算能耗从而减少计算成本。然而，在当前大数据时代背景下情况并不乐观，根据最近的研究报告^[1]显示从2010年到2015年，德国数据中心耗电量增长了15％，增长到12亿千瓦时每年。2014年国际绿色和平组织(GreenpeaceInternational)的报告^[2]称全球数据中心电能的需求量预期到2020年将会增加81％，可见运维数据中心消耗了巨大的电能。通过优化数据中心大数据计算任务调度方式，可以有效地减少大数据计算能耗的开支，从而提高数据中心的能效。

当前，Spark计算框架^[3]被广泛应用于数据中心大数据计算。Spark提供了一个集群的分布式内存抽象即弹性分布式数据集(RDD)，RDD是一个不可变的带分区记录集合，是Spark的编程模型。Spark根据RDD之间的依赖关系把RDD划分成不同阶段(stage)，每个阶段都有一系列方法流水线式地处理RDD。每个阶段内将划分出个数与当前阶段最后一个RDD分区数相同的任务集，也就是一个分区的数据会被一个任务处理。这些任务将调度到worker节点上并行地处理。一个Spark计算框架下大数据应用，在一个worker上只能有一个executor进程，这个进程将占有计算资源，可以处理多个任务。那么任务调度问题就可以描述为装箱问题：任务是物品，executor进程是一些箱子，executor所占有的资源是箱子的大小，任务需要的资源量是物品的大小，将任务分配资源运行，显然这属于NP-hard问题。将任务分配在不同的executor上，运行时间、能耗都不相同，那么合理的分配将对降低大数据应用运行能耗起到至关重要的作用。

目前，Spark自身提供了两种调度策略^[3]，分别是FIFO和FAIR调度策略。这两种调度策略的调度粒度都是阶段。每个阶段内的任务按照本地化原则分配到随机打乱后的executor进程上执行。两种调度策略不同的地方在于：FIFO按照Job生成的顺序和Job下的阶段生成的顺序，为阶段排序，根据排序顺序分配计算资源。FAIR将阶段划分为多个组，每一组称为Pool，按照Pool的权重轮询Pool，为阶段排序，根据排序顺序分配计算资源。杨志伟，郑烇^[4]等人提出一种基于异构Spark集群的自适应任务调度策略。该策略通过监测节点的负载及资源利用率，分析监测得到的参数，自适应动态调整节点任务分配权值，以上调度策略都没有考虑能耗的问题。Leverich和Kozyrakis^[5]提出了一种用于MapReduce作业的能量管理方法。此方法选择性地关闭利用率低的节点降低能耗。他们在覆盖集内至少保留一个数据块副本，关掉不在覆盖集合中且利用率低的节点。L>[6]基于MapReduce计算框架提出了一种高效节能启发式任务调度算法，通过预先对特定作业做能耗分析得到低能耗的任务调度策略，此策略能在保证SLA要求的前提下有效降低能耗。以上节能策略都是基于MapReduce计算框架，目前对Spark能耗研究尚属不足。

Spark的任务调度是一个装箱问题，L>[6]基于MapReduce计算框架提出的启发式任务调度算法，根据预先的能耗分析结果，对Map阶段的计算资源和Reduce阶段的计算资源排序，每个阶段将任务放置在单位能耗低的Slot上。在实际应用中，每次都需要预先对新作业做能耗分析。当集群物理节点发生变化后，原先的性能分析结果需要重新生成，灵活性差，而且不能应用于解决Spark计算框架的能耗问题。

参考文献：

[1]Hintemann R,Beucker S,Clausen J,et al.Energy efficiency of datacenters-a system-oriented analysis of current development trends[C]//Electronics Goes Green.IEEE,2017.

[2]Salahuddin M,Alam K.Information and Communication Technology,electricity consumption and economic growth in OECD countries:A panel dataanalysis[J].International Journal of Electrical Power&Energy Systems,2016,76:185-193.

[3]张安站.Spark技术内幕[M].机械工业出版社,2015.

[4]杨志伟,郑烇,王嵩,等.异构Spark集群下自适应任务调度策略[J].计算机工程,2016,42(1):31-35.

[5]J.Leverich and C.Kozyrakis,On the energy(in)efficiency of Hadoopclusters[J]ACM SIGOPS Oper.Syst.Rev.,vol.44,no.1,pp.61–65,2010.

[6]Mashayekhy L,Nejad M M,Grosu D,et al.Energy-Aware Scheduling ofMapReduce Jobs for Big Data Applications[J].Parallel&Distributed Systems IEEETransactions on,2015,26(10):2720-2733.

[7]罗亮,吴文峻,张飞.面向云计算数据中心的能耗建模方法[J].软件学报,2014(7):1371-1387.

发明内容

本发明旨在解决以上现有技术的问题，提供一种基于能耗感知的大数据能耗模型和基于该模型且能显著降低大数据计算能耗的Spark计算框架调度方法。本发明的技术方案如下：

一种基于能耗感知的Spark节能调度方法，其包括以下步骤：

第一，构建基于Spark计算框架的大数据计算能耗模型；

第二，建立任务与计算资源的能耗关系策略表和执行时间关系策略表，两种策略表共同指导任务调度；

第三，根据策略表选取评价标准最优的计算资源，优先为其分配计算任务，同时保证并行计算任务均衡分配；

第四，通过数据探测初始化决策表，阶段任务执行完后更新能耗关系策略表和执行时间关系策略表。

进一步的，所述大数据计算能耗模型为：大数据应用App产生的能耗为App^ec，

定义如公式(8)所示

降低大数据应用的运行时能耗，即App^ec的值，故目标函数定义如公式(9)所示

其中，i∈Z,j∈Z,k∈Z,l∈Z，表示的任务放置方式，使用表示任务与资源进程ex_l之间的关系，定义如公式(5)所示

当分配到ex_l上时否则

进一步的，所述任务与计算资源的能耗关系策略表：表示Stage_ij集合和Exe的笛卡尔积的运算结果，保存了历史的，任意任务在任意进程(ex_l)上运行时的能耗初始状态，即需要探测有效数据；

任务与计算资源的执行时间关系策略表：任意任务在任意进程(ex_l)上执行的时间初始状态，同样也需要探测有效数据。

进一步的，所述根据能耗关系策略表和执行时间关系策略表选取评价标准最

优的计算资源具体包括：

在一批要并行处理的Stage范围内，对进程ex_l做评价eva_l定义如公式(10)

其中，Stage′表示当前DAGScheduler提交给TaskScheduler处理的Stage，∪Stage′表示TaskScheduler需要处理所有的Stage的并集，∑|Stage′|表示TaskScheduler需要处理所有Stage中task的总数，表示∪Stage′集合中的task^k′在进程ex_l上执行的单位能耗，表示∪Stage′集合中的task^k′单位能耗的总和，按照评价标准结合策略表对进程升序排序，组成队列。

进一步的，当策略表中或者时，表示任务在进程ex_l上的能耗或执行时间为初始化状态，需要将进程ex_l放在队列前面，优先为此进程分配任务。

进一步的，根据进程升序排序队列得到当前最优的进程ex_l，根据执行时间策略表中执行时间的值，将的任务组成Set₀集合，将的任务按照升序组成双端队列TaskQue；

若则优先分配Set₀中的任务到ex_l，直到ex_l资源耗尽或若资源耗尽，则从ExeQue中取出次优的ex_l′继续分配任务；

若即当前决策表数据无需探测或已分配完，将从TaskQue中头尾依次交替取出任务，分配到ex_l，直到ex_l资源耗尽或TaskQue队列为空，若资源耗尽，则从ExeQue中取出次优的ex_l′继续分配任务，若TaskQue队列为空，则已经将所有任务分配结束。

进一步的，当执行时间时，需将此任务单独放入集合Set₀中，优先进行探测。

进一步的，在任务运行结束后，记录本次运行时能耗和运行时间，更新能耗关系策略表和执行时间关系策略表，为下次运行相同任务提供决策。

本发明的优点及有益效果如下：

本发明创新点主要有以下两点：

1.对Spark大数据应用建立能耗模型

本发明首次提出了Spark大数据应用能耗、作业能耗以及阶段能耗的数学模型，详见公式(6)、(7)、(8)，为准确计算spark大数据应用能耗提供模型依据。

优点：能耗模型利用集合的方式描述了，大数据应用(App)与作业(Job)的关系；作业(Job)与阶段(Stage)的关系；阶段(Stage)与任务(task)的关系。使用变量X描述了任务与进程(ex)之间的关系。计算任务在进程上的执行能耗，可灵活地得到阶段能耗、作业能耗以及应用能耗。同时也为Spark能耗研究提供了计算能耗基础。

2.提出能耗感知的Spark节能调度方法

能耗感知的Spark节能调度方法创新点可以细分为

2.1首次提出任务与计算资源的能耗关系策略表和执行时间关系策略表，记录历史的能耗关系、执行时间关系，为任务分配提供策略依据。同时在任务行完后，更新策略表中的数据。优点：使能耗感知的Saprk节能调度方法具有动态扩展性。本调度算法适用于重复运行相同大数据应用的场景，策略表机制在每次运行时动态更新，起到能耗感知的作用。当物理集群变化时，策略表更新机制可以及时感知到未知数据，探测更新。

2.2进程资源依据评价标准排序，用队列保存排序后的结果。这样容易拿到评价标准最优的进程，优先分配任务。评价标准是取当前阶段下的平均能耗。特殊地，需要探测的进程需要放在队列前方。优点：克服了原生Spark调度简单将计算资源随机混洗，不考虑进程资源能耗不同的问题。使本发明提出的节能调度方法有效降低应用运行时的能耗。需要探测的进程需要放在队列前方，保证在物理集群扩展时策略表能动态扩展。

2.3设立Set₀集合保存需要探测的任务，设立TaskQue双端队列保存已知执行时间的任务，按照在进程上的执行时间排序。优先分配Set₀集合中的任务，之后头尾交替分配TaskQue双端队列中的任务。优点：Set₀集合保存的任务优先探测，保证在物理集群扩展时策略表能动态扩展。头尾交替分配TaskQue双端队列中的任务，保证每个进程分配的任务的执行时间均衡，避免出现一个节点运行完后，等待其他节点运行，产生能量消耗的问题。保证了本发明提出的节能调度方法有效降低运行时能耗。

优点：本发明提出能耗感知的Spark节能调度方法，可以有效地减少大数据计算能耗的开支，从而提高数据中心的能效，也可以有效降低由于大数据计算消耗大量电能而造成的温室气体排放量。

附图说明

图1是本发明提供优选实施例的算法图；

图2是计算整体流程图；

图3是MySQL数据库设计E-R图；

图4是WordCount程序逻辑执行图。

具体实施方式

下面将结合本发明实施例中的附图，对本发明实施例中的技术方案进行清楚、详细地描述。所描述的实施例仅仅是本发明的一部分实施例。

本发明解决上述技术问题的技术方案是：

本发明的基本思想是：第一，构建基于Spark计算框架的大数据计算能耗模型；第二，建立任务与计算资源的能耗关系策略表和执行时间关系策略表，两种策略表共同指导任务调度；第三，根据策略表选取评价标准最优的计算资源，优先为其分配计算任务，同时保证并行计算任务均衡分配；第四，通过数据探测初始化决策表，阶段任务执行完后更新能耗关系策略表和执行时间关系策略表。本发明的技术方案包括以下步骤：

步骤一：能耗模型构建

Spark计算框架下，一个大数据应用(App)每遇到action操作会产生一个Job_i，对App定义如公式(1)所示。App表示大数据应用，其由m个Job组成。其中，Job_i根据RDD依赖关系划分为多个Stage_ij，对定义如公式(2)所示。Job_i表示App内第i个Job，其由n个Stage组成。其中，Stage_ij根据最后RDD的分区划分多个对Stage_ij定义如公式(3)所示。Stage_ij表示Job_i上第j个Stage，其由o个task组成。

App＝{Job₀,Job₁,Job₂,…,Job_m-1}>

Job_i＝{Stage_i0,Stage_i1,Stage_i2,…,Stage_i(n-1)}>

其中，0≤i＜m,0≤j＜n,i∈Z,j∈Z

Spark集群内可用的计算资源，即存在于每个worker节点上的executor进程，定义如公式(4)所示。Exe表示集群上所有可用executor进程资源集合。

Exe＝{ex₀,ex₁,ex₂,…,ex_p-1}>

使用表示在ex_l上运行的能耗。使用表示在ex_l上所运行的时间。使用表示任务与资源进程ex_l之间的关系，定义如公式(5)所示

当分配到ex_l上时否则其中，0≤k＜o,0≤l＜p,k∈Z,l∈Z。

根据公式(3)、公式(4)、公式(5)可得Job_i上第j个Stage产生的能耗为定义如公式(6)所示

根据公式(2)、公式(6)可得App内第i个Job产生的能耗为定义如公式(7)所示

根据公式(1)、公式(7)可得大数据应用App产生的能耗为App^ec，定义如公式(8)所示

其中，i∈Z,j∈Z,k∈Z,l∈Z

本发明的目的是降低大数据应用的运行时能耗，即App^ec的值，故目标函数定义如公式(9)所示

其中，i∈Z,j∈Z,k∈Z,l∈Z。表示的任务放置方式，见公式(5)。不合理的任务放置，将会导致obj^ec升高。这就意味着在同样的集群中，运行同样规模的大数据应用，将需要消耗更多的能量。由公式(9)可以看出，如何放置任务，将对目标函数起到决定性作用。

步骤二：策略表初始化

任务与计算资源的能耗关系策略表，定义如表(1)所示。表(1)表示是Stage_ij集合和Exe的笛卡尔积的运算结果，保存了历史的，任意任务在任意进程(ex_l)上运行时的能耗初始状态，即需要探测有效数据。

任务与计算资源的执行时间关系策略表，定义如表(2)所示。表(2)与表(1)不同的地方在于，其保存了历史的，任意任务在任意进程(ex_l)上执行的时间初始状态，同样也需要探测有效数据。

表(1)能耗关系策略表(J)

表(2)执行时间关系策略表(ms)

步骤三：根据评价标准评估单位进程能耗并排序

Spark中，DAGScheduler会将一批能够并行处理的Stage，封装成taskSet送到TaskScheduler进行任务调度，因此本算法在这一批要并行处理的Stage范围内，对任意进程ex_l做评价eva_l定义如公式(10)

其中，Stage′表示当前DAGScheduler提交给TaskScheduler处理的Stage。∪Stage′表示TaskScheduler需要处理所有的Stage的并集。表示TaskScheduler需要处理所有Stage中task的总数。表示∪Stage′集合中的task^k′在进程ex_l上执行的单位能耗。表示∪Stage′集合中的task^k′单位能耗的总和。按照评价标准结合策略表对进程升序排序，组成队列(ExeQue)。特殊地，当策略表中或者时，表示任务在进程ex_l上的能耗或执行时间为初始化状态，需要将进程ex_l放在队列前面，优先为此进程分配任务。

步骤四：将任务优先分配到评价标准最优的进程

由步骤三可以得到当前最优的进程ex_l，根据执行时间策略表将任务按照执行时间升序排序，组成双端队列(TaskQue)。特殊的，当时，需将此任务单独放入集合Set₀中，优先进行探测。

若则优先分配Set₀中的任务到ex_l，直到ex_l资源耗尽或若资源耗尽，则从ExeQue中取出次优的ex_l′继续分配任务。

若即当前决策表数据无需探测或已分配完，将从TaskQue中头尾依次交替取出任务，分配到ex_l，直到ex_l资源耗尽或TaskQue队列为空。若资源耗尽，则从ExeQue中取出次优的ex_l′继续分配任务。若TaskQue队列为空，则已经将所有任务分配结束。

交替分配任务，是为了保证每个进程上分配的任务执行时间尽量平衡，保证不会出现某一台机器执行时间过长，造成其他机器待机，造成不必要的能量消耗。

步骤五：能耗感知的任务调度策略

在任务运行结束后，本算法记录本次运行时能耗和运行时间，更新能耗关系策略表和执行时间关系策略表，为下次运行相同任务提供决策。步骤四的优先探测或者的条件，保证了策略表只要存在未知情况，就会探测出来，保证决策数据完整。本算法协同Spark DAGScheduler完成Spark大数据应用的计算，由步骤一提出的能耗模型，得出大数据应用所产生的能耗。

本发明的实施包括能耗评价模块和调度模块。能耗评价模块对应用程序执行结束后，根据能耗模型得到程序运行能耗，同时兼顾更新策略表的功能。调度模块按照本发明能耗感知调度算法进行任务调度。

1、能耗评价模块

能耗评价模块具有计算worker节点上每个任务线程执行时间的功能，具体做法通过分析Spark提供的监控信息，可以得到任务在进程上的运行时间。也就是说可以得到任意任务在任意进程(ex_l)上执行的时间

能耗评价模块还具有计算worker节点上每个任务线程所产生能耗的功能，具体做法如下：通过操作系统资源监控程序，获得ex_l进程在运行时的CPU使用情况(U_cpu)和内存使用情况(U_memory)。

P＝C₀+C₁×U_CPU+C₂×U_memory>

基于系统使用率的能耗功率模型^[7]，公式(11)所示，可以得到进程在运行情况下的瞬时能耗功率(P_exe)。

由公式(12)得到进程ex_l能耗变化量(P^ec)。P^ec准确反映了负载在进程ex_l上产生的能耗。任务的执行时间与任务的能耗是服从均匀分布的，也就是说任务执行时间越长，任务产生的能耗越大。

由公式(13)可以得到在任意进程(ex_l)上执行的能耗其中∑p^k′l表示在进程ex_l上所有任务执行的时间总和。

能耗评价模块将任务在进程上的执行时间和产生能耗记录在MySQL数据库中，方便调度模块根据记录产生决策，调度任务。数据表设计如图(3)所示。最后，程序执行完成，能耗评估模块根据能耗模型公式(8)得到程序的运行能耗。

2、调度模块

调度模块修改org.apache.spark.scheduler.TaskScheduler的实现类TaskSchedulerImpl。将原始的调度策略粒度从阶段修改为任务，将任务按照执行时间排序。将原始的进程资源混洗，改为按照评价标准排序。将原始的任务按照本地化程度分配，改为按照执行时间分配最优进程。总体来说就是按照Spark能耗感知算法，如图(1)所示，调度根据决策表，调度任务。

下面通过WordCount为例来对本发明进行具体实施说明：

WordCount程序是一个简单统计的单词统计程序。WordCount的主要功能是统计文本中每个单词出现的频次。此程序逻辑执行图，如图(4)所示。其逻辑执行过程主要经过以下几个RDD的转换：

1)从HDFS中读取数据，产生RDD1：textFile。

2)RDD1经过转换操作(transformation)转换为RDD2：flatMap。将单词利用空格分隔，易见RDD2与RDD1是窄依赖关系。

3)RDD2经过转换操作(transformation)转换为RDD3：Map。将单词转换为键值对<word,1>，易见RDD3与RDD2是窄依赖关系。

4)RDD3经过转换操作(transformation)转换为RDD4：reduceByKey。将单词混洗，相同单词的value值相加，易见RDD4与RDD3是宽依赖关系

5)RDD4经过行动操作(action)转换为RDD5：saveAsTextFile。行动操作将触发Job提交，将计算结果保存到HDFS，易见RDD5与RDD4是窄依赖关系

WordCount程序只包含一个Job(Job₀)，Spark>00，Stage₀₁)。RDD4、RDD5划分到Stage₀₀，RDD1、RDD2、RDD3划分到Stage₀₁。根据阶段内最后一个RDD的分区数可以确定Stage₀₀包含2个任务Stage₀₁包含3个任务

假设Spark集群，当前可用的计算资源Exe＝{ex₀,ex₁,ex₂,ex₃}，每个计算资源占有两个CPUCore，即能执行两个任务。已知能耗关系策略表，如表(3)所示，已知执行时间关系策略表，如表(4)所示。

表(3)已知能耗关系策略表(J)

表(4)已知执行时间关系策略表(ms)

调度模块步骤描述如下：

提交Stage₀₁阶段

1)提交executor排序生成ExeQue队列

由公式(10)得到executor的评价{eva₀＝0.84,eva₁＝2,eva₂＝1.17,eva₃＝1.32}则ExeQue＝{ex₀,ex₂,ex₃,ex₁}

2)取出最优进程，所有任务排序TaskQue双端队列

取出进程ex₀，按照在ex₀上执行时间排序，则

3)分配任务

先分配到ex₀，即再分配到ex₀，即任务没有分配完，ex₀资源耗尽，取出次优进程ex₂。分配到ex₂，即

4)任务执行完后，由能耗评价模块计算任务的执行时间和能耗，记录数据到MySQL数据库中。

提交Stage₀₀阶段

1)对executor排序生成ExeQue队列

由公式(10)得到executor的评价值{eva₀＝0.5,eva₁＝3,eva₂＝0.85,eva₃＝1.83}则ExeQue＝{ex₀,ex₂,ex₃,ex₁}。

2)取出最优进程，所有任务排序TaskQue双端队列

取出进程ex₀，按照在ex₀上执行时间排序，则

3)分配任务

先分配到ex₀，即再分配到ex₀，即

4)任务执行完后，由能耗评价模块计算任务的执行时间和能耗，记录数据到MySQL数据库中。

本例只包含两个阶段，Spark DAGScheduler在所有阶段计算完后，即本Job计算结束。本例中程序只包含一个Job，所以，程序结束。能耗评价模块给出程序的能耗值与运行时间。

任务调度结果：

Stage₀₀阶段，分配到进程ex₀，分配到进程ex₀。

Stage₀₁阶段，分配到进程ex₀，分配到进程ex₀，分配到进程ex₂。

可以根据表(3)和表(4)以及公式(8)得到app^ec＝(1+3)+(3+6+10)＝23。以上这些实施例应理解为仅用于说明本发明而不用于限制本发明的保护范围。在阅读了本发明的记载的内容之后，技术人员可以对本发明作各种改动或修改，这些等效变化和修饰同样落入本发明权利要求所限定的范围。