Storm环境下基于权重的任务调度算法

鲁 亮，于 炯，卞 琛，英昌甜,3，师康利，蒲勇霖

(1.新疆大学 信息科学与工程学院，乌鲁木齐 830046； 2.新疆大学 软件学院，乌鲁木齐 830008;3.新疆大学 电气工程学科博士后科研流动站，乌鲁木齐 830047)

0 引言

随着互联网和各类智能终端的普及，数据呈现出井喷式发展的趋势，MapReduce等各类大数据处理框架应运而生[1-2]。然而，这类传统的大数据批量处理框架无法满足部分企业的实时性业务需求。Apache Storm[3-4]作为一个开源、实时、分布式部署、容错且扩展性良好的大数据流式计算系统[5-6]，已成功解决这一问题并引起了学术界和企业界的高度关注。在Storm系统中，只要数据源处于活动状态，元组便会源源不断地发送至各工作节点，计算和传输将持续发生，无需进行中间结果的持久化存储，在实时个性化推荐、实时交通大数据分析、实时临床数据分析等领域具有广阔的应用前景[7-9]。

Storm在进行任务分配时采用轮询(Round-Robin，RR)调度算法，即将用户提交的拓扑中包含的每一个任务均匀分配到各工作进程中，再将各工作进程均匀分配到各工作节点上，未考虑到各任务计算开销的差异以及任务与任务之间不同类型的通信开销，这将对Storm处理的实时性产生较大影响。针对这一问题，已有少量国内外学者展开相关研究。文献[10]提出资源感知的在线调度算法R-Storm，将Storm资源分为硬约束(针对内存)和软约束(针对CPU和网络)两类，利用任务需求的各类静态资源和工作节点所能提供的静态资源之间的关系实现调度，最终达到最大化资源利用率和提高集群吞吐量的效果，但该算法中各任务的资源需求完全依靠程序员人为设定而并非通过监测获得，不适合数据流快速变化场景下的在线调度。文献[11]对此进行改进，添加了资源负载监测模块并将监测结果存入数据库，并使用调度生成器根据数据库中的数据进行实时调度，但并未评估调度自身对系统运行带来的影响以及重调度后是否能够带来更低的系统延迟。文献[12]提出Storm框架下流量感知的在线调度算法T-Storm，通过监测任务负载、工作节点负载以及任务与任务之间的数据传输负载，将通信开销大的任务动态分配至空闲资源较多的节点上。然而，该算法并无法保证通信开销大的一对任务一定分配到同一个节点中，且一个拓扑需求的节点数量依赖于用户设定。此外，文献[13]提出一种带权图的k划分算法，文献[14-16]提出流式计算框架下实时高效的资源调度算法和优化框架。以上研究解决了流式计算环境下的任务优化调度问题，但无法直接移植于Storm平台。文献[17]提出并实现了一种服务质量感知的Storm分布式调度器，其在网络时延和可靠性等方面均优于Storm默认调度算法的执行结果，但这与本文集中式数据中心的研究背景不符。文献[18]将拓扑热边的概念引入Storm平台，其主要思想是将热边关联的源任务和目标任务调度到同一工作节点执行，以达到减少网络通信开销的效果。然而该算法未充分考虑拓扑内部通信的全局性，并未将非热边的调度优化考虑在内。文献[19]为了分别降低进程间通信开销和节点间通信开销，在充分考虑Storm拓扑中各任务通信情况的基础上提出一种两阶段调度算法，但该算法并未考虑到各任务自身计算开销的差异性。文献[20]提出Storm环境下离线和在线的两种自适应调度算法，其中离线调度算法用于拓扑运行前的拓扑结构分析，而在线调度算法用于拓扑运行中各节点的实时负载监测和任务分配的动态调整，其中在线调度算法效果更优。这种自适应调度算法解决了部分Storm环境中的通信开销问题，但执行时仅逐对考虑相互通信的任务，未将当前任务与其直接通信的所有任务全部考虑进来，这对于较为复杂的拓扑而言尚缺乏全局性，容易陷入局部最优，且当所有任务间的数据传输速率一致时，该算法无法工作。

为了改进Storm默认调度算法、克服已有相关研究的不足，本文提出一种Storm环境下基于权重的任务调度算法(Task Scheduling Algorithm based on Weight in Storm，TSAW-Storm)。该算法根据实时监测到的负载数据，为拓扑中包含的任务和数据流分别赋予不同的点权和边权，并根据本文提出的负载均衡模型和最优通信模型的要求，将任务调度到合适的工作节点内。实验结果表明，相对于Storm默认调度算法和文献[20]的在线调度算法，TSAW-Storm在Storm集群系统延迟、通信开销和负载均衡方面均有所改进。

1 Storm作业模型

在Storm中，一个流式作业称为一个拓扑，表示为一个有向无环图，由组件和数据流共同构成。组件分为Spout和Bolt两种：其中Spout为数据源编程单元，用于为拓扑运行提供数据；Bolt为数据处理编程单元，用于实现拓扑中的处理逻辑。数据流是由无限的元组组成的序列，是Storm中对传递的数据进行的抽象，可通过不同的流组模式实现组件之间的数据流传输。如图1所示，ca和cb为数据源编程单元Spout，其余组件为数据处理编程单元Bolt，所有箭线为数据流；特别地，组件cf和cg为数据终点，通常用于将最终数据展示至终端或持久化至数据库。

图1 拓扑逻辑模型 Fig. 1 Logical model of a topology

为提高拓扑执行的并行度，每个组件均可同时运行多个实例，称之为任务。任务是各组件最终执行的代码单元。设tij表示组件ci运行的第j个实例，sij,kl表示任务tij向任务tkl发送的数据流。数据流通过描述拓扑中数据的流向，将上游任务和下游任务关联起来，由此提出如下关联任务的概念。

定义1 关联任务。对于任意任务tgh、tij与tkl，若存在任务tgh向任务tij发送的数据流sgh,ij与任务tij向任务tkl发送的数据流sij,kl，则任务tgh与tkl统称为任务tij的关联任务。

图2即为图1的一种实例模型。特别地，当组件ci的实例数量为1时，定义ti1=ti。可以看到cd的实例数量为3，cf的实例数量为2，其余组件的实例数量为1。以任务td1为例，其上游任务ta与tb以及下游任务tf1与tf2均称为任务td1的关联任务。

图2 拓扑实例模型 Fig. 2 Instance model of a topology

在Storm集群中，资源池由一系列工作节点构成，定义该集合为N={n1,n2,…,n|N|}。每个工作节点内配置有若干个槽，每个槽只能容纳一个工作进程。文献[12]通过Storm吞吐量测试[21]表明，对于单个拓扑而言，若在一个工作节点内分配多个工作进程(即占用多个槽)，将会增加进程间通信开销进而导致运行效率的下降。因此，本文仅在一个工作节点内分配一个工作进程，此时Storm默认的轮询调度算法可简化为一个拓扑中包含的所有任务在各工作节点上的均匀分配，任务与工作节点间的对应关系可定义如下。

定义2 任务分配法则。若任务tij分配到了工作节点nk上，则记f(tij)=nk或f-1(nk)=tij；若任务集合Tnk={t11,t12,…,tij,…}分配到了工作节点nk上，则记f(Tnk)=nk或f-1(nk)=Tnk，其中f即为任务或任务集合在工作节点上的分配法则。如图3为图2的拓扑运行于包含有3个工作节点的Storm集群中的任务分配模型，若以工作节点n1及运行在该工作节点上的任务集合为例，可表示为：f({td1,tc,tf1})=n1，f-1(n1)={td1,tc,tf1}。

图3 任务分配模型 Fig. 3 Model for task assignment

由图3可知，Storm系统中的任务之间存在两种通信模式，分别是类似于任务td1与tf2之间的节点间通信以及类似于任务td1与tf1之间的节点内通信。其中节点间通信受制于当前环境下的网络带宽和带宽利用率大小，开销往往很大；而节点内通信与网络无关，由于一个任务运行于一个工作线程内，因此其通信类型仅为进程内的线程级通信，开销很小并可忽略不计[14]。然而Storm默认轮询的任务调度算法并未兼顾到这两种通信模式的差异性，从而导致较大的节点间通信开销。此外，由于各组件自身业务逻辑的不同，CPU占用率必然存在差异；即使是同一个组件中各个业务逻辑相同的任务，也将由于部分流组模式(如Field Grouping)的限制，并不会将其接收到的所有数据流均分到各实例中，故各任务的计算开销依然存在差异，轮询的任务分配方式无法保证集群中各工作节点的负载均衡。为了更好地评估任务的计算开销和任务间传输的数据流大小，提出如下带权拓扑的概念。

定义3 带权拓扑(Weighted Topology, WT)。设任务tij占用的CPU资源大小为wtij，任务tij与tkl之间传输的数据流大小为wsij,kl，则图2的拓扑实例模型中各任务和任务间的数据流可分别使用wtij和wsij,kl构成拓扑的点权和边权，其值可通过后文提出的负载监视模块进行实时获取。这样的拓扑实例模型称作带权拓扑。

2 问题建模与分析

本节在带权拓扑的基础之上建立负载均衡模型与最优通信模型。其中负载均衡模型建立了各工作节点理想负载与实际负载间差异大小的度量指标，可为调度算法运行时的终止条件和运行后的负载均衡效果评估提供理论依据；最优通信模型则证明了节点间数据流传输代价与节点内数据流传输代价此消彼长的转化关系，为最小化通信开销的任务迁移过程提供了决策原则。

2.1 负载均衡模型

设某一带权拓扑中包含的任务集合为T，nx为工作节点集合N中任意一个工作节点，工作节点nk上分配的任务集合为f-1(nk)，工作节点nl(k≠l)上分配的任务集合为f-1(nl)，则必然有：

(1)

且：

f-1(nk)∩f-1(nl)=∅;k≠l

(2)

记Wnx为工作节点nx的CPU负载，其值为分配给工作节点nx上的所有任务需要占用的CPU资源总量，即:

(3)

记W为Storm集群的CPU负载总和，即集群中各工作节点的CPU负载总量。在同构环境下，各工作节点的CPU负载随任务分配模型的变化而此消彼长，但W始终不变，其计算方法为:

(4)

若将集群中的CPU负载总和均匀分布到各工作节点上，则每个工作节点需容纳的CPU负载为:

(5)

式(5)表示在理想状态下达到负载均衡后各工作节点的CPU负载情况。事实上，集群中各工作节点不可能达到完全的负载均衡状态，故使用式(6)标准差来衡量各工作节点的实际负载与理想负载的偏离程度。标准差越小则表示各工作节点间的负载差异越小，负载越为均衡。

(6)

2.2 最优通信模型

如第1章所述，Storm系统中任务之间的通信模式可分为节点间通信和节点内通信，其中节点间的通信开销远大于节点内的通信开销。因此，若寻求Storm系统中通信模式的优化途径，需令节点间通信开销达到最小，即尽可能减少工作节点间传输的数据流总和，可表示为:

(7)

定理1 最优通信开销原则。最小化节点间传输的数据流大小等价于最大化节点内传输的数据流大小，即式(7)等价于:

(8)

证明 由第1章Storm作业模型可知，拓扑一旦提交到集群，拓扑实例模型即固定下来，其包含的任务总数和数据流总数不可改变。因此在不发生拥塞的情况下，总数据流大小为一定值C，即:

(9)

证毕。

由定理1可知，在进行Storm系统任务调度时，为了达到最优通信模型的要求，应尽可能地把通信频繁的任务调度到同一个工作节点上，以最大限度地降低节点间通信开销。

3 基于权重的任务调度算法

为了达到上述负载均衡模型和最优通信模型的要求，本章提出Storm环境下基于权重的任务调度算法(TSAW-Storm)，并进行算法评估与部署。

3.1 边权增益

TSAW-Storm旨在尽可能均衡分配各工作节点负载的前提下，尽量减少节点间传输的数据流总和。而由最优通信模型可知，当任务分配法则发生变化时，节点间数据流与节点内数据流将相互转化。为了量化这一过程，提出如下边权增益的概念。

定义4 边权增益。对于任务tij，若存在任务分配法则f(tij)=np，设Ttij,np为与任务tij关联且位于工作节点np上的任务集合，Ttij,nq为与任务tij关联且位于工作节点nq上的任务集合，且p≠q，若将任务tij由工作节点np迁移至工作节点nq，则边权增益可表示为：

(10)

以图3为例，对于任务td3而言，位于该任务所在工作节点n3上的关联任务集合为Ttd3,n3={ta}，位于工作节点n2上的关联任务集合为Ttd3,n2={tb,tf2}，若将任务td3由工作节点n3迁移至工作节点n2，则数据流sb,d3和sd3,f2将由节点间数据流转化为节点内数据流，而数据流sa,d3将由节点内数据流转化为节点间数据流，其边权增益的计算方法为：

Gtd3=wsb,d3+wsd3,f2-wsa,d3

(11)

可见，边权增益反映了任务迁移前后节点间数据流的差值。边权增益越大，则意味着将更多的节点间数据流转化为了节点内数据流，这符合2.2节中最优通信模型的要求，是后续算法的设计目标之一。

3.2 算法描述

计算边权增益的前提是获取带权拓扑的关联任务及其对应边权；而为了在获取最大化边权增益的同时不违背负载均衡模型的约束，带权拓扑的点权及各工作节点的CPU负载值同样不可或缺。因此在TSAW-Storm的设计过程中，需在用户提交拓扑后首先执行Storm默认调度算法，并当拓扑运行稳定后完成上述数据的采集和存储。当工作节点的CPU负载持续不均，即集群中所有工作节点的最大CPU负载与最小CPU负载之差在时间间隔τ内持续大于阈值ε时，则触发算法1进行任务的重新调度。具体步骤如下：

算法1 TSAW-Storm。

2)对于∀nx∈N′，如果nx中存在拓扑运行所需要的数据源且n|N|中分配的任务包含有Spout实例，或nx中不存在数据源但n|N|中分配的任务不包含有Bolt实例，则随机将工作节点n|N|中的一个Spout实例迁移至工作节点nx；否则随机将工作节点n|N|中的一个Bolt实例迁移至工作节点nx。设从工作节点n|N|中迁出的任务为tij，则有Tnx←{tij}，Tn|N|←Tn|N|-{tij}，同时根据式(3)更新工作节点CPU负载Wn|N|和Wnx。

3)获取与Tnx中各任务关联且位于工作节点n|N|上的任务集合，记为Tnx,n|N|。

4)将Tnx,n|N|中边权增益最大的任务tkl迁移至工作节点nx，此时Tnx←Tnx∪{tkl}，Tn|N|←Tn|N|-{tkl}，同时根据式(3)更新工作节点CPU负载Wn|N|和Wnx。

6)重复第2)～5)步，逐步构建N′中各工作节点上分配的任务集合。

7)执行任务分配，即对于x∈1,2,…,|N|，实施f-1(nx)←Tnx。

3.3 算法解释与算法评估

在算法1的执行过程中，第1)步获取算法后续计算所需数据，并进行各项初始化操作。第2)步完成当前工作节点上的第一个任务分配，这是进行边权增益计算的前提。在第一个任务的选择过程中，如果待迁入任务的工作节点上存储有数据源，则将拓扑中的数据源编程单元Spout分配到该工作节点上，目的是尽量避免Spout将远程数据源读入拓扑时带来的节点间通信开销，提高任务本地化执行的概率，进而提高Storm系统的运行效率。然而，将Spout中包含的所有任务均分配到数据源所在工作节点上的做法是不可取的，原因有以下两点：其一，拓扑中Spout的各个任务彼此之间并无关联，若将其分配到同一个工作节点上，势必导致更小的节点内数据流，与最优通信模型的要求不符；其二，Spout中的每一个任务均需读取其下游Bolt运行所需的所有数据，输出的数据流总量必然很大，若各任务的分布过于集中，势必给其所在工作节点与其下游Bolt所在工作节点之间带来过多的节点间数据流，进而导致网络拥塞，效果适得其反。为了避免上述情况的发生，第2)步仅为存在数据源的每个工作节点初始化分配一个Spout实例。第3)～6)步严格遵循负载均衡模型和最优通信模型的要求，根据最大化边权增益原则，逐步选择当前情况下最合适的任务从工作节点n|N|上迁出，并及时更新任务迁移后相关工作节点CPU负载的预测值，最终确定所有工作节点上的任务分配法则。这一过程存在遍历关联任务、计算最大边权增益等需要较大时间开销的重复性工作，时间复杂度为O(|S|·|N|)(S为带权拓扑包含的数据流集合)；然而以上步骤实质上并未改变任务分配模型，拓扑依然正常运行，用户的实时性作业需求并未受到影响。直到步骤7)时才执行具体的任务分配，此时需重新分配拓扑中各任务在各工作节点中的映射关系，时间开销为O(|T|)；在这一时刻，拓扑执行会不可避免地存在短暂的中断，延迟将有所增加，具体情况将在第4章实验中进行评估。

3.4 算法实现与部署

为实现并部署算法1，需使用Storm为开发人员提供的可插拔自定义任务调度器，即实现接口org.apache.storm.scheduler.IScheduler中的方法public void schedule(Topologies topologies, Cluster cluster)。改进后的Storm架构如图4所示。需要说明的是，一个完整的Storm分布式系统由运行进程Nimbus的主控节点、运行进程Supervisor的工作节点、运行进程UI的控制台节点以及运行进程ZooKeeper的协调节点共同构成，而图4并未修改控制台节点和协调节点的工作机制，故将其省略，仅保留新增模块以及与新增模块相关联的部分，其中新增加的四个模块如下：

1)负载监视模块。负责在一定的时间窗口内，收集各任务占用的CPU负载信息及各任务之间的数据流大小分别作为带权拓扑的点权和边权。由于Storm中的一个任务运行于一个工作线程中，因此为了获取任务执行过程中占用的CPU资源大小以及各对任务在单位时间传输的元组数量，需实时追踪各任务对应的线程ID及其相关联的所有线程。其中各线程的CPU资源占用大小可通过ThreadMXBean类的getThreadCpuTime(long id)方法获取到其占用的CPU时间，并与其所处工作节点的CPU主频相乘求得；各对线程间传输的数据流大小需使用计数器变量统计各线程接收到的上游线程发送的元组数量，并与时间窗口容量相除获得数据流传输速率。具体实现需添加在组件中各Spout的open()和nextTuple()方法以及各Bolt的prepare()和execute()方法中。

2)MySQL数据库。负责存储历次任务分配信息以及负载监视模块传来的负载信息和数据流大小，并实时更新。

3)任务调度模块。负责读取数据库中的信息，并在负载持续不均时触发算法1以及时作出任务调度决策。

4)自定义调度器。覆盖主控节点中Storm默认轮询的调度算法，负责读取任务调度模块生成的调度决策并执行任务调度。

图4 改进的Storm系统架构 Fig. 4 Improved architecture of Storm

4 实验与分析

4.1 实验环境与分析

实验环境采用相同硬件配置的PC搭建一个包含有12个节点的Storm集群，其中共同运行进程Nimbus、进程UI和数据库服务MySQL的节点1个，运行进程ZooKeeper的协调节点3个，其余8个为运行进程Supervisor的工作节点。表1列出了各节点具体的软硬件配置。

表1 Storm集群软硬件配置Tab. 1 Hardware and software configuration of Storm cluster

实验基于Intel公司Zhang[22]发布在GitHub上的基准测试storm-benchmark-master，本文选取其中CPU敏感型(CPU-Sensitive)的WordCount构建拓扑，数据源为其自身提供的原版英国历史小说《双城记》，格式为txt。原著中各单词出现的频率不尽相同，因此在实际生产环境中具有一定的代表性，若此时使用Storm默认轮询的调度算法，极易在计数过程中发生各工作节点CPU负载不均的情况，进而触发TSAW-Storm，便于评估算法的执行效果。表2列出了WordCount运行时的各项参数配置及其相关说明。需要进一步解释的是，表2中工作进程个数设置为8，意味着8个工作节点中各分配1个工作进程，这样便消除了同一节点内工作进程间的通信开销，与文中任务分配模型的描述相符；Acker用来跟踪元组的处理结果，其值默认设置与工作进程个数相同；Spout缓存队列长度可对Spout的元组发射速率进行控制，并进而决定Storm系统的吞吐量，通过多次实验后设置该集群配置下的合适值为50；时间间隔tau和阈值epsilon即为3.2节中叙述的τ与ε，表示若集群中所有工作节点的最大CPU负载与最小CPU负载之差在每80 s时间间隔内持续超过20%时，触发TSAW-Storm，该值可根据Storm默认调度算法的运行结果进行人为调整。

表2 WordCount参数配置Tab. 2 Parameter configuration of WordCount

为验证TSAW-Storm的有效性，文中除了与Storm默认调度算法进行对比之外，还部署了文献[20]的Storm自适应在线调度算法(online scheduler)。该算法在运行后取得了较好的调度效果，其核心思想是实时监测各工作节点和各任务的CPU负载情况以及各对任务之间的数据流大小，当存在CPU负载持续超出阈值的工作节点时触发任务重分配机制，即首先按照递减的顺序排列拓扑中各对任务之间的数据流大小，然后将任务逐对调度至那些令其重分配后产生最低CPU负载的工作进程和工作节点中。表3列出了采用在线调度算法进行对比实验时的各项参数配置。需要说明的是，为了与TSAW-Storm在同等CPU负载条件下触发任务调度，各项参数均通过多次实验进行微调并最终确定了理想值，其中参数reschedule.timeout为在线调度算法的触发周期，参数capacity为在线调度算法中CPU的使用率上限，这两项参数分别与本文算法中的τ与ε存在对应关系。

表3 在线调度算法参数设置Tab. 3 Parameter configuration of online scheduler

4.2 实验结果与分析

本节通过实验评估TSAW-Storm在系统延迟、通信开销和负载均衡三个方面的表现。为便于数据统计，以下各项测试均在基准测试的配置文件中设置metrics.poll为5 000，metrics.time为300 000，其单位为ms，即每组实验每5 s进行一次采样，总时长为5 min。

4.2.1 系统延迟测试

延迟表明一个元组从Spout发射到最终被成功处理的时间消耗，反映了拓扑执行一次的响应时间，刻画了系统的运行效率。图5统计了WordCount在Storm默认调度算法(图例中Default)、在线调度算法(图例中Online)与基于权重的任务调度算法(图例中TSAW-Storm)下的系统延迟。

图5 三种任务调度算法下的系统延迟对比 Fig. 5 Comparison of latency among three task scheduling algorithms

如图5所示，从实验开始到第一个峰值结束时间段表示拓扑提交时的初次任务分配，此时的任务分配均遵循Storm默认调度算法。其中0～25 s的零延迟阶段表示任务分配的计算与同步过程，由于此时存在未被成功调度的任务，拓扑实例模型并不完整，故无法形成一条完整的数据流；同时，Spout中的任务往往首先完成分配并开始发射数据，在其下游Bolt中的任务未被成功调度的情况下，Spout缓存队列中的元组因无法及时得到处理而导致系统延迟随着运行时间的推进而不断增加，进而出现30～35 s的极高峰值。第一个峰值过后，系统延迟逐渐趋于收敛，在集群中各工作节点不发生意外的情况下，默认调度算法将不再实施任务调度，系统延迟的保持在11.2 ms左右；而此时在线调度算法与TSAW-Storm开始收集集群中各工作节点以及工作节点上各任务占用的CPU负载信息和各任务之间的数据流大小，为各任务未来的优化配置提供决策依据。

随着运行时间的推移，第90 s时在线调度算法触发，此时所有任务在各工作节点上重新分配，系统暂停一切数据流传输，故系统延迟在90～100 s时间间隔内无法观测；第105～110 s时系统延迟达到极高峰，随后迅速下降，整个任务重调度过程相当于拓扑提交时的初始化任务分配，执行开销较大。在线调度算法触发的原因是此时集群中已经存在CPU负载在80 s内超过70%的工作节点，而之所以图5中第90 s才出现系统延迟的极端变化，其原因有以下几点：1)主控节点的重调度指令分发到各工作节点需要消耗一定的时间；2)调度发生时，Spout虽然不再发射数据，但整个拓扑中仍存在少量未被完全处理的数据流，Acker机制仍在进行系统延迟的统计工作；3)采样周期为一定值，统计误差不可避免。由图5可知，在线调度算法运行时对系统的影响范围在第90～115 s，最大延迟为91.9 ms；系统运行稳定后，延迟平均值约为8.50 ms，相对于Storm默认调度算法降低约24.1%。

TSAW-Storm的触发和执行过程与在线调度算法类似。第90 s时TSAW-Storm触发，原因是在80 s的观测周期内存在最大CPU负载与最小CPU负载之差持续大于20%的工作节点。与在线调度算法不同的是，TSAW-Storm触发后的系统延迟在90～95 s时间间隔内无法观测，仅为在线调度算法的一半左右，且随后发生的延迟最高峰值为27.5 ms，仅为在线调度算法最大延迟的29.9%，可见TSAW-Storm对Storm系统的正常运行并未造成较大的影响。TSAW-Storm运行结束后，系统运行迅速趋于稳定，延迟平均值稳定在约7.84 ms，相对于在线调度算法降低约7.76%，相对于Storm默认调度算法降低约30.0%。TSAW-Storm触发时之所以对系统整体影响较小，是因为该算法首先由主控节点计算更优的任务分配方案，这一过程并未改变任务分配模型，拓扑依然正常运行；而后再根据计算结果一次性执行任务分配，故影响系统正常运行的时间很短， Spout缓存队列中的元组等待处理的时间也较短，不会导致类似在线调度算法产生的突发延迟。而TSAW-Storm之所以在收敛后能够形成较低的系统延迟，是因为该算法不同于在线调度算法中逐对任务调度的方法，它针对带权拓扑中的每一个任务，均充分考虑了与其相关联的所有数据流，其调度更具全局性。此外，TSAW-Storm提高了任务本地化执行的概率，消除了一部分Spout中的任务读取远程数据源时的网络开销，这是导致系统延迟降低的又一个重要原因。

4.2.2 通信开销测试

本节讨论基准测试WordCount在Storm默认调度算法、在线调度算法与TSAW-Storm下的工作节点间通信开销。图6为采用三种不同调度算法时工作节点间单位时间的元组传输总量。

图6 三种任务调度算法下的节点间数据流大小对比 Fig. 6 Comparison of inter-node tuple rate among three task scheduling algorithms

与图5中的系统延迟类似，图6中节点间数据流大小的统计结果亦可清晰反映在线调度算法与TSAW-Storm的触发情况。可以看出，无论是采用Storm默认调度算法的初始化任务分配，还是在线调度算法与TSAW-Storm触发后的优化调度，节点间数据流大小均将经历一个从0迅速上升到正常状态的过程，并不存在类似图5中的峰值情况。这是因为表2中对WordCount的Spout缓存队列长度作了合理限制，当未被成功处理的元组达到缓存队列的上限时，Spout将暂停发射数据流，因此并不会发生因元组大量累积而突发传输的情况。由图6可知，Storm默认调度算法执行且系统运行趋于稳定后(50～300 s)，节点间数据流大小的平均值约为92 446 tuple/s；而当在线调度算法和TSAW-Storm触发且系统稳定运行后(分别为125～300 s和115～300 s)，节点间数据流大小的平均值约分别为70 335 tuple/s和62 026 tuple/s，相比Storm默认调度算法分别降低了23.9%和32.9%，其中TSAW-Storm相比在线调度算法执行后的节点间数据流大小下降了11.8%。可见，TSAW-Storm在降低节点间通信开销方面具有更为明显的效果，其原因是TSAW-Storm中最大化边权增益和Spout任务本地化的思想能够最大范围地考虑到整个带权拓扑的任务间通信情况，从而将更多的节点间数据流转化为节点内数据流。而之所以TSAW-Storm在4.2.1节中降低的系统延迟不如降低的节点间数据流大小效果明显，是因为在各类Storm基准测试中，WordCount属于CPU敏感型拓扑[22]，节点间数据流的减小仅可作为该类拓扑性能优化的方向之一，未来将针对拓扑的自身特性探索更多可能的优化方向。

4.2.3 负载均衡测试

本节讨论基准测试WordCount在Storm默认调度算法、在线调度算法与TSAW-Storm下分别运行时集群的负载均衡情况。由于Storm默认调度算法采用轮询的方式分配任务，因此各工作节点上初始化分配的任务数量相同。然而WordCount数据源中各单词出现的频率存在很大差异，当SplitBolt采用Field Grouping方式进行数据流分发时，各CountBolt中的任务需要处理的数据流大小差异很大，单纯采用轮询方式进行任务分配极易导致各工作节点的负载不均。在线调度算法与TSAW-Storm在任务重调度过程中充分考虑到不同任务负载的差异性，从而克服了默认调度算法在负载均衡方面的不足。表4为采用这三类任务调度算法执行任务分配后各工作节点的CPU负载均值。

由于在线调度算法和TSAW-Storm触发前均使用Storm默认调度算法执行任务分配，因此可将表4中的Storm默认调度算法(Default)执行后的CPU负载看成是在线调度算法(Online)和TSAW-Storm触发前各工作节点的资源占用情况。由表4可知，Storm默认调度算法执行后，各工作节点的负载不均现象较为严重，标准差高达12.66%，其中6号工作节点的CPU负载最高，7号工作节点的CPU负载最低，二者差值为33.4%。因最大CPU负载超出表3中设置的CPU使用率上限(capacity=70%)，且最大CPU负载与最小CPU负载之差也超出了表2中设置的阈值(epsilon=20%)，故在线调度算法和TSAW-Storm同时触发，这与图5中系统延迟以及图6中节点间数据流大小的统计结果也是相吻合的。在线调度算法和TSAW-Storm执行后，集群中各工作节点的CPU负载均低于70%且负载基本均衡，不必再次触发任务调度；标准差分别为3.47%和3.26%，仅为Storm默认调度算法的27.4%和25.8%。可见两种任务调度算法均能达到负载均衡的效果，且TSAW-Storm效果略优，其执行后的CPU负载标准差相比在线调度算法降低了5.93%。这是因为在向除最后一个工作节点之外的其他工作节点逐步迁入任务的过程中，每个工作节点实际容纳的负载大小均不能超过其理想情况下的CPU负载，这比在线调度算法中每次寻找具有最低CPU负载的工作节点的做法在负载均衡方面更便于控制。然而通过表4可以发现，使用TSAW-Storm执行任务调度后，8号工作节点的CPU负载将略低于其他7个工作节点，这是由于该调度算法的工作机制导致的：在算法1的初始化过程中，带权拓扑中的所有任务都被拟分配至8号工作节点，而后再结合Spout任务本地化和最大化边权增益的思想，将8号工作节点中的任务逐一分析并迁移至其他7个节点，直到各工作节点中分配的负载大小均在刚好大于其理想情况下的CPU负载为止。这种做法虽然不易导致节点过载，且能够较好地保证1～7号节点的负载均衡，但可能导致位于8号工作节点上的任务被过多地迁出，进而发生该工作节点的CPU负载略低于其他工作节点的情况。若需解决这一问题，可在TSAW-Storm执行结束后，在满足最优通信模型的前提下进行少量任务交换，这将在未来继续开展研究。

表4 三种任务调度算法下的CPU负载对比Tab. 4 Comparison of CPU load among three task scheduling algorithms

5 结语

Storm默认轮询的任务调度算法并未考虑到各任务计算开销的差异以及任务之间不同类型的通信模式，在负载均衡和节点间通信开销方面仍存在较大的优化空间。针对这一问题，本文将各任务占用的CPU资源大小作为拓扑的点权，任务间的数据流大小作为拓扑的边权，提出带权拓扑的概念；并在此基础上建立负载均衡模型和最优通信模型，进而提出Storm环境下基于权重的任务调度算法(TSAW-Storm)。该算法利用最大化边权增益的思想逐步构建起各工作节点中承载的任务集合，在保证集群负载均衡的同时，尽可能将节点间数据流转化为节点内数据流，从而减小网络开销，提高Storm系统的运行效率。实验通过基准测试WordCount从系统延迟、通信开销以及负载均衡三个方面论证了本文调度算法的有效性。

下一步研究工作主要集中在以下几个方面：1)本文实验开展的背景为同构集群环境，下一步研究中将探索拓扑权值与CPU性能和网络带宽的关系，将TSAW-Storm移植至异构Storm集群环境下；2)根据TSAW-Storm调度后的任务分配结果进一步尝试优化，如采用任务交换等方法，解决某一工作节点负载较低的问题；3)从拓扑自身的结构特征出发，将TSAW-Storm进一步推广至更为复杂的Storm商业应用领域，使其适用于更为丰富的业务场景。

参考文献(References)

[1] 孟小峰,慈祥.大数据管理:概念、技术与挑战[J].计算机研究与发展,2013,50(1):146-169. (MENG X F, CI X. Big data management: concepts, techniques and challenges [J]. Journal of Computer Research and Development, 2013, 50(1): 146-169.)

[2] CHEN C L P, ZHANG C Y. Data-intensive applications, challenges, techniques and technologies: a survey on big data [J]. Information Sciences, 2014, 275(11): 314-347.

[3] TOSHNIWAL A, TANEJA S, SHUKLA A, et al. Storm @Twitter [C]// Proceedings of the 2014 ACM SIGMOD International Conference on Management of Data. New York: ACM, 2014: 147-156.

[4] Apache. Apache Storm [EB/OL]. (2017-08-01) [2017-08-10]. http://storm.apache.org.

[5] 孙大为,张广艳,郑纬民.大数据流式计算:关键技术及系统实例[J].软件学报,2014,25(4):839-862. (SUN D W, ZHANG G Y, ZHENG W M. Big data stream computing: technologies and instances [J]. Journal of Software, 2014, 25(4): 839-862.)

[6] RANJAN R. Streaming big data processing in datacenter clouds [J]. IEEE Cloud Computing, 2014, 1(1): 78-83.

[7] 王铭坤,袁少光,朱永利,等.基于Storm的海量数据实时聚类[J].计算机应用,2014,34(11):3078-3081. (WANG M K, YUAN S G, ZHU Y L, et al. Real-time clustering for massive data using Storm [J]. Journal of Computer Applications,2014,34(11):3078-3081.)

[8] 乔通,赵卓峰,丁维龙.面向套牌甄别的流式计算系统[J].计算机应用,2017,37(1):153-158. (QIAO T, ZHAO Z F, DING W L. Stream computing system for monitoring copy plate vehicles [J]. Journal of Computer Applications, 2017, 37(1): 153-158.)

[9] TA V D, LIU C M, NKABINDE G W. Big data stream computing in healthcare real-time analytics [C]// Proceedings of 2016 IEEE International Conference on Cloud Computing and Big Data Analysis. Piscataway, NJ: IEEE, 2016:37-42.

[10] PENG B Y, HOSSEINI M, HONG Z H, et al. R-Storm: resource-aware scheduling in Storm [C]// Proceedings of the 16th Annual Middleware Conference. New York: ACM, 2015: 149-161.

[11] 刘月超,于炯,鲁亮.Storm环境下一种改进的任务调度策略[J].新疆大学学报(自然科学版),2017,34(1):90-95. (LIU Y C, YU J, LU L. An improved task schedule strategy in Storm environment [J]. Journal of Xinjiang University (Natural Science Edition), 2017, 34(1): 90-95.)

[12] XU J L, CHEN Z H, TANG J, et al. T-Storm: traffic-aware online scheduling in Storm [C]// Proceedings of the 34th IEEE International Conference on Distributed Computing Systems. Piscataway, NJ: IEEE, 2014: 535-544.

[13] 郑丽丽,武继刚,陈勇,等.带权图的均衡k划分[J].计算机研究与发展,2015,52(3):769-776. (ZHENG L L, WU J G, CHEN Y, et al. Balancedk-way partitioning for weighted graphs [J]. Journal of Computer Research and Development, 2015, 52(3): 769-776.)

[14] SUN D W, ZHANG G Y, YANG S L, et al. Re-Stream: real-time and energy-efficient resource scheduling in big data stream computing environments [J]. Information Sciences, 2015, 319: 92-112.

[15] GHADERI J, SHAKKOTTAI S, SRIKANT R. Scheduling storms and streams in the cloud [J]. ACM SIGMETRICS Performance Evaluation Review, 2015, 43(1): 439-440.

[16] LIU Y, SHI X, JIN H. Runtime-aware adaptive scheduling in stream processing [J]. Concurrency and Computation: Practice and Experience, 2016, 28(14): 3830-3843.

[17] CARDELLINI V, GRASSI V, LO PRESTI F, et al. Distributed QoS-aware scheduling in Storm [C]// Proceedings of the 9th ACM International Conference on Distributed Event-Based Systems. New York: ACM, 2015: 344-347.

[18] 熊安萍,王贤稳,邹洋.基于Storm拓扑结构热边的调度算法[J].计算机工程,2017,43(1):37-42.(XIONG A P, WANG X W, ZOU Y. Scheduling algorithm based on Storm topology hot-edge [J]. Computer Engineering, 2017, 43(1):37-42.)

[19] ESKANDARI L, HUANG Z, EYERS D. P-Scheduler: adaptive hierarchical scheduling in Apache Storm [C]// Proceedings of the Australasian Computer Science Week Multiconference. New York: ACM, 2016: Article No. 26.

[20] ANIELLO L, BALDONI R, QUERZONI L. Adaptive online scheduling in Storm [C]// Proceedings of the 7th ACM International Conference on Distributed Event-Based Systems. New York: ACM, 2013: 207-218.

[21] MARZ N. Public stormprocessor/storm-benchmark [EB/OL]. (2012- 08- 20) [2017- 08- 10]. https://github.com/stormprocessor/storm-Benchmark.

[22] ZHANG M. Intel-hadoop/storm-benchmark forked from manuzhang/storm-benchmark [EB/OL]. (2015- 11- 02) [2017- 08- 10]. https://github.com/intel-hadoop/storm-benchmark.

This work is partially supported by the National Natural Science Foundation of China (61462079, 61562086), the Natural Science Foundation of Xinjiang Uygur Autonomous Region of China (2017D01A20), the Educational Research Program of Xinjiang Uygur Autonomous Region of China (XJEDU2016S106), the Graduate Research and Innovation Project of Xinjiang Uygur Autonomous Region of China (XJGRI2016028).

LULiang, born in 1990, Ph. D. candidate. His research interests include distributed computing, in-memory computing.

YUJiong, born in 1964, Ph. D., professor. His research interests include grid computing, distributed computing.

BIANChen, born in 1981, Ph. D., associate professor. His research interests include distributed computing, in-memory computing.

YINGChangtian, born in 1989, Ph. D.. Her research interests include in-memory computing, green storage.

SHIKangli, born in 1990, M. S. candidate. Her research interests include distributed computing, in-memory computing.

PUYonglin, born in 1991, M. S. candidate. His research interests include in-memory computing, green computing.