0 概述

物联网、社交网络、云计算等技术的出现，同时伴随计算能力、空间存储和网络带宽的快速发展，人类生活中的各类数据在互联网、通信、金融、商业、医疗等诸多领域不断增长和累积。由于数据量日益增多，并且在各种应用场景下业务需求和竞争压力对数据处理的实时性和准确性提出了更高的要求，因此传统的单机器数据处理技术已经无法应对时代的变化，而大数据的发展面临着现实应用的难题。因此，各种创新性的技术得到了充分发展，如云计算^[1]、量子计算、分布式计算等技术。其中，分布式计算以其强大的计算能力和低廉的成本得到了广泛应用和深入研究。分布式系统^[2]通常分为批数据处理系统和流数据处理^[3]系统两类，批处理应用广泛的有Hadoop^[4]和Spark^[5]等，流处理应用广泛的有Storm^[6]、Flink^[7]、Spark streaming^[5]等。相对于批处理系统，流处理更关注数据的实时性。实时数据流具有高速、不稳定的特征，这也导致使用分布式流处理系统时对于计算资源必须有合理而准确的调度。

分布式集群资源的动态调度是当前的一个研究热点，如Hadoop自带的YARN^[8]以及Apache开源的Mesos^[9]框架，它们都可以在集群中对资源进行管理。而在国内，也有类似于基于分簇的小基站用户资源分配方案^[10]，但不足之处在于，这种管理更多地需要人工干预，无法实时监测集群资源的利用情况并进行合理的资源分配。在流数据处理任务中，集群需要一直处于准备状态，以应对流数据处理负载的变化。因此，需要根据负载情况，通过负载均衡对任务进行合理分配，如文献[11]提出基于负载平衡和经验值的工作流任务分配策略。而对于流数据处理任务，资源负载是实时变化的，因此，如何在不停变化的实时负载过程中自动地对集群资源进行整理，降低当前暂不需要的资源的消耗，是亟需解决的一个难题。

资源动态调度及协同分配的实现，难点在于如何根据当前系统的资源使用情况对后续资源需求做出准确的预测，以及对预测的资源计算出当前对系统加权延迟最低的资源分配比例，同时动态关闭未使用的资源，减少资源消耗。本文通过对流数据处理任务负载进行实时监测，构建一个双层资源调度模型，利用ZooKeeper^[12]和YARN进行动态资源管理，从而实时改变集群资源分布情况，减少系统延迟。

1 系统构建背景

在资源量化的方式上，本文系统是在YARN的基础上构建的，而在YARN中，资源被虚拟化为容器这个概念，每个容器资源中包含了相对应的存储空间和CPU大小等资源。

在资源动态调度方面，本文将资源调度的处理分为任务层和系统层两个层面。在任务层，文献[13]提出基于排队论的动态资源调度模型，该模型需要在集群大小固定的情况下进行调度。以此为基础，本文提出在资源调度时根据实时负载而改变集群大小减小延迟的模型。在系统层，已知的资源调度框架有YARN和Mesos等，主要解决了物理资源调度、将物理资源映射为虚拟化资源以及将虚拟化资源分配给分布式流处理系统进行使用这3个问题。但关于分布式流数据系统如何将不同的资源分配给不同的任务则没有涉及。

在Hadoop自带的YARN资源管理框架中有3种资源调度算法，即先来先服务调度算法、公平调度算法和Capacity调度算法。这3种算法在资源调度上基本可以实现对资源的合理分配以应对各种情况。但是YARN的调度算法是基于人为改变资源需求的前提，无法对资源的需求进行预测和动态调整。针对上述问题，本文提出一种双层资源调度模型，通过中间件沟通系统层和YARN资源调度框架对集群资源进行动态调度和协同分配处理。

在集群管理方面，已知的如Storm on YARN^[14]、Flink on YARN^[15]、Storm on Mesos^[16]、Flink on Mesos^[17]等开源框架，都只是实现了对固定大小集群在多任务处理时的集群管理，但框架仍然需要更多人力进行管理，在资源负载变化时人为改变集群的大小。

综上所述，本文系统以DRS动态资源调度模型^[13]作为任务层，用以监测实时延迟和负载变化等情况，并通过系统层沟通集群资源，通过增减资源实现对分布式流数据处理任务的动态资源调度管理和协同分配。

2 基于Storm on YARN与DRS的分布式系统 2.1 Storm on YARN框架

Storm on YARN是Yahoo开源的一个基于Storm的分布式集群资源管理框架^[18]，为基于Storm的分布式流数据处理任务提供了较完善的资源管理机制，其主要由ResourceManager、ApplicationManager、NodeManager、Container、Nimbus、Supervisor等部分组成。Storm on YARN系统结构如图 1所示。

Storm on YARN系统主要有以下优点：

1）弹性计算资源。在将Storm运行在分布式集群^[19]中时，Storm可以和其他的应用程序（例如flink流数据处理任务或Spark批处理^[20]任务）共享集群资源，当Storm流数据处理任务负载较大时，可以动态地为其增加资源；当Strom流数据处理任务负载较小时，可以为其减少资源，并将资源给予其他需要的应用程序。

2）共享底层存储。YARN可以为集群中运行的分布式框架提供共享的HDFS^[21]存储，减少管理多个集群带来的不便以及传输数据带来的时间延迟。

3）多版本共存。不同版本的Storm都可以运行在一个YARN上，降低了版本维护的成本。

当然，Storm on YARN本身还可以做出一些改进，目前的Storm on YARN依然需要人为地通过对分布式流数据处理任务负载的监测，再借由人为操作对集群资源进行管理，这样当集群较大时，无疑对管理集群来说是一个十分巨大的挑战，如何在多个流数据处理任务情况下根据当前任务的重要程度协同分配资源，是亟待解决的问题。本文立足于此设计解决方案。

2.2 DRS系统

DRS是一个分布式动态资源调度系统，该系统基于Storm分布式流数据处理框架，通过排队论对流数据处理的时间延迟进行处理和预测，从而对集群资源的分配做出合理分配的一个集群资源管理系统。DRS系统工作流程如图 2所示。

DRS系统工作具体步骤如下：

步骤1  系统收集预设时间窗口内的负载信息，包括$ {\lambda }_{(\mathrm{s}\mathrm{o}\mathrm{u}\mathrm{r}\mathrm{c}\mathrm{e}, m)}\mathrm{、}{\lambda }_{(i, \mathrm{i}\mathrm{n})}\mathrm{、}{\lambda }_{(i, \mathrm{o}\mathrm{u}\mathrm{t})}\mathrm{、}{\mu }_{i}\mathrm{、}{s}_{i}\mathrm{、}{k}_{i}\mathrm{、}{p}_{ji} $等。

步骤2  数据拟合还原，将收集到的时间窗口内的数据通过预设的数据拟合函数进行拟合，使之后对资源调度的决策更接近真实数据结果。

步骤3  将拟合后的数据提交到资源调度器以供其做出正确的决策。

在DRS系统中，资源调度器对分布式流数据处理任务的负载情况界定包括3种状态，即合适（feasible）、过载（overproviding）和资源不足（shortage）。DRS根据资源调度器对负载情况做出的负载情况界定，并通过时间窗口内通过数据拟合后得到的数据，对分布式流数据处理任务后续需要分配的资源情况做出正确的决策，包括删减资源或增加资源等。但DRS的不足之处在于，它只对固定集群大小的资源进行分配，无法控制集群本身做出关闭节点减少资源或打开节点提供资源等操作。

3 双层资源调度模型原理及实现

由于DRS资源调度器只能在固定给定物理资源的条件下对资源进行合理分配和调度，而在分布式流处理任务中，有时根据任务的负载情况需要减少集群资源或者增加集群资源，而这需要通过人为帮助才能做到，即系统层资源调度器无法通过任务层资源调度器的分配策略动态调整集群资源情况。

本文提出双层资源调度模型，该模型结合了任务层资源调度器和系统层资源调度器，并在系统层通过Zookeeper与YARN建立连接，通过YARN对集群资源进行调整。

如图 3所示，双层调度模型分为任务层资源调度器和系统层资源调度器，任务层资源调度器的作用是管理集群的所有工作任务，根据当前各个任务的负载情况，以最有效降低整体任务延迟为目的，制定出符合当前情境的资源分配策略。例如，在当前集群中，假设各节点的资源占用为（5，5，5），此时节点1的延迟高于预设值，而节点3的延迟低于预设值，则任务层资源调度器可能会将资源占用比设定为（6，5，4）。任务层资源调度器使用文献[9]基于排队论的动态资源调度模型，用于制定合理的资源分配策略。系统层资源调度器的作用为依赖Storm on YARN框架对集群资源的高级管理能力，根据中间件调度器给出的资源信息增加或减少资源。

此过程相对目前的Storm on YARN框架而言，能够起到动态分配资源的作用，而不需要人为增减资源，并且由于根据中间件调度器的能力来控制资源总数，对于当然资源冗余的节点，可以实时地关闭节点，不会占用集群资源，方便其他集群使用。

3.1 双层资源调度模型的工作流程

本文提出的双层资源调度模型主要服务于分布式流数据处理任务过程，其对集群资源进行增减管理，并在多个任务中考虑到不同任务的重要性不同，根据任务权重协同地调配资源比例。该模型工作的具体步骤如下：

步骤1  任务层资源调度器对分布式流数据处理任务，根据当前负载做出决策。

步骤2  任务层资源调度器发出资源调度请求，由中间件调度器接收。

步骤3  对于集群目前的资源情况，中间件调度器做出决策，当资源冗余时，跳至步骤4，当资源不足时，跳至步骤5，否则跳至步骤6。

步骤4  当出现资源冗余时，中间件调度器与系统层资源调度器发起联系，并按照资源使用情况，优先关闭资源使用率较低的节点，然后通知中间件调度器，完成流程。

步骤5  当出现资源不足时，如果通过打开集群中关闭节点可满足需求时，则中间件资源调度器直接与系统层资源调度器沟通；否则，中间件调度器根据预设的任务权重，由计算出的多个任务的权重延迟比，对资源进行协同分配，再通过与系统层资源调度器沟通，分配资源。

步骤6  中间件资源调度器检测到目前资源为最优状态，触发流数据任务的资源调整。

3.2 中间件调度器的增减资源策略

基于双层资源调度模型的工作流程，本文主要关注点在于中间件调度器在不需要协同分配资源时对集群节点的减增策略。

设$ \{{S}_{0}^{\mathrm{o}\mathrm{l}\mathrm{d}}，{S}_{1}^{\mathrm{o}\mathrm{l}\mathrm{d}}，\cdots ，{S}_{N-1}^{\mathrm{o}\mathrm{l}\mathrm{d}}\} $为当前时刻分布式集群占用资源情况，其中，$ {\mathrm{S}}_{i}^{\mathrm{o}\mathrm{l}\mathrm{d}} $表示处理节点i当前时刻的资源情况。同理，设$ \{{S}_{0}^{\mathrm{n}\mathrm{e}\mathrm{w}}, {S}_{1}^{\mathrm{n}\mathrm{e}\mathrm{w}}, \cdots , {S}_{N-1}^{\mathrm{n}\mathrm{e}\mathrm{w}}\} $为任务层资源调度器决策后得到的分布式流处理任务目标资源情况。设N₁为决策后需求节点数，N₂为集群当前拥有节点数，设R为增减算法结果，中间件调度器的资源增减算法描述如下：

算法1  中间件调度器资源增减算法

输入  $ \{{S}_{0}^{\mathrm{n}\mathrm{e}\mathrm{w}}，{S}_{1}^{\mathrm{n}\mathrm{e}\mathrm{w}}，\cdots ，{S}_{N-1}^{\mathrm{n}\mathrm{e}\mathrm{w}}\} $，$ \{{S}_{0}^{\mathrm{o}\mathrm{l}\mathrm{d}}，{S}_{1}^{\mathrm{o}\mathrm{l}\mathrm{d}}，\cdots ，{S}_{N-1}^{\mathrm{o}\mathrm{l}\mathrm{d}}\} $

输出  R

1.If $ \{{\mathrm{S}}_{0}^{\mathrm{n}\mathrm{e}\mathrm{w}}，{\mathrm{S}}_{1}^{\mathrm{n}\mathrm{e}\mathrm{w}}，\cdots ，{\mathrm{S}}_{\mathrm{N}-1}^{\mathrm{n}\mathrm{e}\mathrm{w}}\} $≠$ \{{\mathrm{S}}_{0}^{\mathrm{o}\mathrm{l}\mathrm{d}}，{\mathrm{S}}_{1}^{\mathrm{o}\mathrm{l}\mathrm{d}}，\cdots ，{\mathrm{S}}_{\mathrm{N}-1}^{\mathrm{o}\mathrm{l}\mathrm{d}}\} $

2.Rebalance

3.Else

4.Return R=true

5.End if

6.If N₁ < N₂

7.SubSlots

8.Else if N₁=N₂

9.Break

10.Else

11.AddSlots

12.End if

13.DefaultSecheduler

14.Return R=true

在算法1中，CallRebalance指Storm改变本身的资源分配平衡，之后中间件调度器察觉到分布式流处理任务所需资源发生变化，则与当前系统所有资源做对比，并通过与系统层资源调度器的会话联系，做出相应的资源调整，增加或者删减资源，最后中间件调度器察觉到当前资源处于可提供的最优状态，触发流处理任务的资源调整。具体步骤如下：

步骤1  需要任务层资源调度器预先生成调度决策。生成资源调度决策包含信息监测、诊断生成和决策生成3个部分。其中，系统运行时状态收集是第一步，也是很重要的一步，只有收集到足够多、足够准确的系统状态，才能做出正确的资源调度决策。而其中对于数据堆积情况的监测尤为重要，因为在实时系统中，稳定运行状态下一旦发生数据堆积，系统实时性会降低，延迟会增大。此时，必然是因为工作负载变大或者系统本身出现问题，应当迅速对此做出反馈，解决数据堆积的情况。例如，在判断数据堆积的情况时，可以通过基于排队论的方法^[9]来实时地反映系统不断变化的状态。最后再利用决策算法^[9]，通过多次迭代来保证资源分配比例最优。

步骤2  在系统层资源调度器层面，将系统可利用的物理资源虚拟化为可分配、可量化的虚拟化资源，如容器节点，然后根据中间件调度器的指示进行动态地增加或者删减资源，目的是自动对资源进行管理，减少人工介入。

在上述步骤中，当中间件调度器与系统层资源调度器会话时，如果当前集群可启动的节点无法满足所需资源，则会给中间件调度器返回false，之后中间件调度器会一直尝试获得资源，直到对资源需求改变，或者需求得到满足。

3.3 中间件调度器的协同分配资源策略

在任务层资源调度器中，本文通过预设的时间窗口将分布式流处理任务的实际延迟保存到一个预设长度的队列中，根据该队列中的数据计算出预设的拟合函数的参数，并将参数发送到中间件调度器。在中间件调度器中，利用DRS中的排队论算法^[9]，根据流处理任务的目标节点数获得改变资源后的预测延迟，再通过拟合函数拟合出更接近真实数据的值。

当多个任务在系统中并且集群无法满足这些任务所需要的的资源时，需要权衡各个任务的重要性，从而协同地分配资源。通过将获得的预测延迟（已拟合过）与预设的每个任务的权重做加权延迟和，并比较不同资源分配比例时哪个加权延迟和最小，将该结果为最佳资源协同分配方案。

在需要协调分配资源，即有多个流数据处理任务要同时管理（目前只支持两个）时，将为不同的任务设定不同的权重来代表该任务的重要性，每个任务在任务层资源调度器中，依然通过3.2节中的调度决策策略来根据自身负载改变资源分配比例。由于此时有多个任务需要改变自身节点数，因此中间件调度器会先向系统层查询当前集群的资源数是否满足任务层资源需求。例如，任务1的资源需求由（5，5，5）- > （5，5，6），任务2的资源需求由（5，5，5）- > （5，6，6），此时中间件调度器会向系统层查询集群资源是否满足，由于系统资源只能提供30个节点资源，因此任务1和任务2需要协同分配资源。中间件调度器根据任务层资源调度器的反馈，依据两个任务的权重不同，通过穷举的方式计算出各种分配方案下系统总延迟最佳的分配方案，从而将新的分配方案通知给系统层进行资源分配。

4 系统验证 4.1 实验目的

通过实验验证本文系统的资源动态调节功能和多任务协同分配功能是否运行正常。

在由虚拟机搭建的Hadoop集群上进行YARN资源调度相关实验。通过对虚拟机配置的统一性，排除在实验过程中由于机器配置如内存、CPU、网络等因素对实验结果造成的不必要的影响。相关配置如表 1所示。

4.2 实验设计

本文基于YARN对Storm分布式流数据处理任务进行动态资源调度和协同分配。通过一个简单的单词统计（以下简称WordCount）任务来验证系统功能是否正常，之所以只通过单词统计来进行实验，是为了方便在实验过程中对任务负载通过句子发送频率进行调控。

实验中WordCount任务以一个包含100 000条句子的文件作为数据源，通过采用Redis做消息队列，向Storm的spout以预设的速率发送句子。在Storm的完全延迟趋于稳定后，通过改变消息队列的发送速率来调整分布式流处理任务的负载情况，从而使系统对资源的需求发生改变。WordCount任务拓扑如图 4所示。可以看出，在WordCount中，当切分单元、统计单元和报告单元对当前到达的数据的速率（工作负载）λ_i超过各个单元本身算子对数据的处理速率μ_i乘以单元个数k_i时，表明该单元的资源数需要调整。

在介绍关于分布式流处理任务的负载改变如何控制后，设置如表 2所示的一组实验参数。其中，数据源的输入速率λ_s=400 tuple/s，而此时设置的切分单元的处理能力为k_切分单元×μ_切分单元=2×150=300 tuple/s。若切分单元可以处理所有到达的句子，在本文实验数据集中，每个句子平均可以切分为6个单词，则下一单元到达率为6×400=2 400 tuple/s，此时统计单元的处理能力为k_统计单元×μ_统计单元=5×400=2 000 tuple/s < 2 400 tuple/s。此外，报告单元的处理能力为k_报告单元×μ_报告单元=5×400=2 000 tuple/s，也小于报告单元数据的到达率。系统将根据当前的负载情况，判断出此时资源处于shotage状态，从而重新做出新的资源分配决策，中间件调度器检测到分布式任务的资源请求发生改变，则会在通过计算当前资源是否能满足任务需求后，再决定是否向系统层资源调度器申请资源。

以上为检测系统的资源动态调度功能，而对于系统多任务系统分配资源的功能，本文则通过设置2个相同的分布式流处理任务，在2个任务同时改变任务负载的情况下，判断当前的集群资源分配是否可协同分配，根据权重和预测延迟的改变做出更合理的分配决策。

4.3 实验结果分析

验证系统的功能是否工作正常，主要关注2个指标的变化：一个是在负载变化后，集群是否做出决策；另一个是在集群资源重新分配后，分布式流处理任务的延迟是否发生改变。实验以关闭动态调度功能的系统作为对比，验证系统功能。

如图 5所示，基于Storm on YARN的动态资源调度功能在最初消息队列发送数据不稳定，因此，本文过滤掉前1 min的latency数据，之后在第3 min，任务层资源调度器做出了对资源分配的新的决策，即（2，5，5）- > （3，6，6），此时集群资源足够对任务分配做出响应，但集群只使用了6台共12个节点，所以，需要中间件调度器向系统层资源调度器申请资源，之后资源分配变为（3，6，6）共15个节点，需要使用8台机器，latency降到预设的范围，并趋于稳定。

如图 6所示，在上一个实验的基础上，通过Strom提供的Rebalance机制，强制调整分布式流数据处理任务WordCount的节点分配为（1，4，4），可以看出，任务节点分配改变后，WordCount任务的延迟突增，这是因为任务资源无法满足当前的数据输入速率，此时任务层资源调度器会重新做出资源分配决策，中间件调度器察觉到任务层资源调度器做出决策后，则提供动态调度功能调整集群资源分配，之后任务延迟回到正常水平。

如图 7所示，设置WordCount_1的节点分配为（3，6，6），在有两个相同分配的WordCount任务同时进入集群时，由于WordCount_1先进入集群，因此其占据了15个节点数，此时另一个任务WordCount_2只能获得节点分配为（1，2，2），设置2个任务的权重都为0.5，则在中间件调度器为任务2修改资源时，会根据2个任务的权重公平地分配资源，所以，此时2个节点的资源分配都是（2，4，4）。

以上3个实验都验证了本文系统具备动态调度系统资源的功能，在通过动态调度后，任务的延迟可以达到预设的延迟，提高流数据处理任务的处理效率。

相比于非动态调度，双层资源调度模型可以分层准确地对系统进行信息监测，诊断生成及生成决策，并通过中间件调度器的调度，将分配策略同步给系统层资源调度器，从而对系统可分配的虚拟化资源进行合理分配，降低系统延迟，这也是双层调度框架相对于非动态调度可以提高系统资源利用率的原因。

5 结束语

本文提出一种基于YARN的分布式流数据处理任务资源的动态调度系统。相对于开源Storm on YARN框架，该系统基于双层调度模型，可检测流数据处理任务延迟，实现对集群负载的动态调度，从而实时调整集群大小，避免人工干预集群，降低集群维护成本。本文系统只能对2个任务进行协同分配，下一步将在协同分配多个任务资源方面对其进行改进，并设计只通过穷举方式计算资源分配的方案。