亓开元;韩燕波;赵卓峰;马强
【摘 要】为了在大规模历史感知数据基础上实现针对高速传感数据流的实时计算,提出一种面向大规模历史数据的数据流处理方法RTMR,通过中间结果缓存、流水化和本地化改进了MapReduce的数据流处理能力.在此基础上,为了适应性地构造RTMR集群,利用模型分析方法根据应用特征和集群环境配置节点类型和拓扑结构.为实现集群的负载均衡,通过计算负载状态转换关系分组空闲节点和过载节点,将NP难的动态负载均衡问题快速分解为规模较小的子问题,并且综合执行时间和数据移动代价作为子问题的优化目标,提高应对负载倾斜的反应速度.实验表明,上述方法和技术能够保障大规模历史数据上数据流处理的可伸缩性.%With the
development of Internet of Things, how to realize real time computation for high speed data stream based on large scale history sensor data became a new challenge to cloud manufacturing. A processing method named Real-Time MapReduce (RTMR) oriented to large scale historical data was proposed, which improved data stream processing capacity of MapReduce through intermediate result cache, pipelining and localization. To construct RTMR sets, the model analysis method was used to configure the node type and topological structure based on application
characteristics and cluster environments. Furthermore, to realize cluster load balancing, the idle nodes and overload nodes were grouped by computing load state transition relatioa Thus the dynamic load balancing problem of NP hard was decomposed into small scale sub-problems, and execution time as well as data cost were integrated as sub-problem's
optimization objective. The experiment result showed that the proposed method and technology could ensure the scalability for data stream processing of large scale historical data. 【期刊名称】《计算机集成制造系统》 【年(卷),期】2013(019)003 【总页数】13页(P641-653)
【关键词】数据流处理;大规模数据处理;MapReduce方法;适应性架构;负载均衡 【作 者】亓开元;韩燕波;赵卓峰;马强
【作者单位】北方工业大学云计算研究中心,北京100144;中国科学院计算技术研究所,北京100190;中国科学院大学,北京100190;北方工业大学云计算研究中心,北京100144;北方工业大学云计算研究中心,北京100144;中国科学院计算技术研究所,北京100190;中国科学院大学,北京100190 【正文语种】中 文 【中图分类】TP393
1 问题的提出
借助物联网、云计算等技术实现制造资源和制造能力的共享、协同与服务化是云制造的基本追求。随着物联网的发展，以实时传感数据为基础的各类数据流处理逐渐成为当前云制造资源感知和接入的主要形式。面对连续的传感数据流，处理系统必须快速响应并及时输出结果。然而有限的处理系统不可能处理无限数据流的完整信息，因此一般采用窗口机制来划定处理边界，边界范围内的数据称为感知历史数据。
随着数据采集和传输技术的进步，数据流速度不断提高，使得短时间内积累大量的历史数据成为可能。同时，当前数据流处理的长期性和准确性需求也要求扩大历史数据规模。以一个物联网环境下的车辆监管系统为例，该系统通过传感设备收集城市车辆运行数据，并实时完成数据流同历史数据的统计、分析、比较等计算，从而实现套牌、超速、限行等违法车辆的自动识别。在这类应用中，窗口边界范围的不断扩大、数据处理对象（如车牌）数量的急速增加，以及每个处理对象数据量（如车辆监控信息）的迅速增加，导致历史数据规模不断扩大。在上述趋势下，如何在面对大规模历史数据时保证数据流处理的实时性，即满足数据流处理对历史数据的伸缩性需求，成为云制造领域的新挑战。
现有数据流处理领域对于可伸缩性的研究分为集中式和分布式两类。在集中式的单节点环境下，受内存容量限制，侧重于针对小规模的历史数据进行处理，主要通过准入控制［1］、服务质量（Quality of Service，QoS）降阶［2］、概要数据［3］等方法，以牺牲服务质量为代价提供伸缩性。在分布式集群环境下，针对由多个数据流处理算子组成的处理网络，主要通过在集群节点上分布算子来提供伸缩性［1－2］，但处理能力仍局限于单个算子所在节点所能处理的窗口范围，在历史数据规模不断扩大的情况下伸缩能力不足。
为了突破单个节点的计算能力和内存容量限制，支撑大规模数据的处理和存储，当前的数据中心广泛采用多核集群计算架构，以及Cache、内存、外存和分布式存储四层存储结构，如图1所示。在这种架构中，节点上的多核CPU构成了本地计算资源；节点上的内外存组成了本地存储。在无共享架构下，MapReduce［4］编程模型通过简单的编程接口为并行处理大规模数据提供了充足语义，向程序员屏蔽了任务调度、数据存储和传输等细节，已经被广泛接受。然而，现有的 MapReduce方法，如 Hadoop［5］，Phoenix［6］都属于对持久化数据的静态批处理方式。以这种方式处理持续到达的数据流，若每次处理小批的数据，则系
统开销太大，实时性受到限制，主要表现在：①在每次处理时都要初始化运行环境，重复地载入和处理历史数据；②Reduce阶段要等到Map阶段完成后才能进行，阶段间存在同步关系；③在节点间需传递大量中间结果数据。若等待批达到一定规模又增加了处理延迟，则同样无法满足实时需求。因此，为满足面向大规模历史数据的实时流处理的需求，如何扩展MapReduce模型是需要考虑的问题。
针对制约MapReduce流处理能力的重复处理、阶段同步和数据传输等瓶颈，可以通过缓存中间结果避免历史数据的重复处理开销，并使得数据流处理流水化和本地化，减少阶段间的同步开销和节点间的数据传输开销。基于该技术路线，本文基于理论分析提出了一种支持此类数据处理的方法RTMR（real－time MapReduce），并在此基础上着重考虑以下问题：
（1）从RTMR实现角度（节点类型配置、集群拓扑结构等）选择多种组合来构造集群架构，针对不同数据流处理应用和集群环境，如何适应性地构造集群架构是一个需要解决的问题。
（2）从RTMR集群角度，节点上的负载倾斜是影响整体吞吐量、制约数据流处理可伸缩性的主要因素。因此，如何实现负载均衡也是保障数据流处理可伸缩性的关键问题。
针对上述问题，本文通过模型分析，提出一种适应性的RTMR集群构造技术，并从静态数据分布和动态负载调整两方面考虑，提出了一种负载均衡算法。 2 面向大规模历史数据的数据流处理方法
利用MapReduce模型满足大规模数据处理的伸缩性需求是云制造的一项核心技术，然而现有基于批处理方式的MapReduce方法无法满足高速数据流下的大规模数据实时处理需求。本章针对制约MapRduce流处理能力的瓶颈因素，通过理论分析给出其可利用缓存、流水化和本地化机制进行实时性改进的理论基础，并提
出了一种面向大规模历史数据的数据流处理方法。 2．1 RTMR方法
MapReduce模型的定义［4］为
其处理过程是：Map方法将〈k1，v1〉键值对转换为〈k2，v2〉键值对，Reduce方法针对每个k2 的值列表List〈v2〉做list操作。若待处理数据为D，Map阶段中间结果为I，M 表示 Map方法，R表示Reduce方法，则上述过程可以表示为MR（D）＝R（M（D））＝list（I）。为了分析 MapReduce的性质，给出如下定义：
定义1 对于函数F：I→O，若存在函数P：O×O→O，使得F（D＋Δ）＝P（F（D），F（Δ）），则称F为可合并。
定义2 对于数据集D的n个数据子集D1，D2，…，Dn，若D1∩D2∩…∩Dn＝∅且D1∪D2∪…∪Dn＝D，则称D1，D2，…，Dn 为D 上的划分。
定义3 对于键值对数据集合D＝｛〈key，value〉｝，键集合K，称集合｛d｜d．key∈K，d∈D｝为D在K 上的选择，记为σK（D）。 根据上述定义可知，MapReduce具有以下性质：
（1）Map方法满足分配率，即两个数据集合并集上的Map等于分别对两个集合Map的并集
（2）Reduce方法具有可合并性，即list（D＋Δ）＝list（list（D），list（Δ））。
（3）Reduce方法满足分配率，即若 K1，K2，…，Kn为中间结果I键集合的一个划分，则list（I）＝list（σK1（I））＋list（σK2（I））＋…＋list（σKn（I））。
在批处理方式的MapReduce中，大规模历史数据的重复处理开销是制约数据流处理能力的关键因素。为了利用预处理和中间结果缓存改进MapReduce的数据流处理能力，证明其具有可合并性。 定理1 MapReduce可合并。
证明 根据MapReduce的性质，对于数据D和增量Δ，有
因此，由定义1可知MapReduce是可合并的。
定理1表明可通过预处理并缓存中间结果减少每次数据流到达时的历史数据重复处理开销，提高MapReduce的实时处理能力。将上述过程表示为MR（D＋Δ）＝list（ID＋IΔ）＝MR（Δ｜ID）。
在现有的MapReduce方法中，阶段间的同步开销也是制约数据流处理能力的主要因素。同步的产生原因是Reduce阶段等待所有Map任务完成并对其结果进行合并排序。实际上，定理1同时表明Map和Reduce阶段间并不存在数据依赖关系，因此可以利用异步的流水方式消除阶段间的同步。在Map和Reduce阶段间建立缓冲区，每个Map任务在处理完成后直接将结果放入缓冲区，Reduce任务从缓冲区取得数据进行处理。此外，节点间的数据传输开销也是制约MapReduce数据流处理能力的重要因素，因此应充分利用本地计算资源完成处理，减少数据传输。下面证明MapReduce可以利用本地化技术改进数据流处理能力。
定理2 K1，K2，…，Kn为中间结果I键集合的一个划分，对于数据增量Δ在I上的 MapReduce，有
证明 由MapReduce的性质，对于中间结果I和数据增量Δ，有
对于Reduce方法来说，在中间结果的K1选择上处理的数据增量仅与K1有关，
即
定理2表明在集群上分布缓存中间结果可以使数据流同历史数据的MapReduce处理仅发生在节点本地。因为避免了节点间的数据传输，所以合理地划分中间结果能够保障集群的可伸缩性。如何按照节点能力划分中间结果，并在运行时动态地进行调整，将在第5章介绍。
根据上述理论分析，设计了面向大规模历史数据的数据流处理方法RTMR，其工作过程为（如图2）：
（1）预处理历史数据生成中间结果，根据k2的Hash值划分，分布缓存到各个工作节点的本地存储上。
（2）MapReduce以异步流水方式进行，各节点Map阶段按照Hash区间划分将数据流路由到相应的Reduce节点进行与中间结果的计算。 （3）将本地计算结果更新到分布式存储。
在RTMR方法中，工作节点负责维护本地中间结果缓存和阶段流水线，控制节点负责RTMR作业的生命周期管理、可靠性和可伸缩性保障。本文主要对RTMR可伸缩性保障中的架构构造和负载均衡方法进行讨论。 2．2 中间结果缓存和阶段流水线
为减少每次数据流到达时的重复计算开销，RTMR支持中间结果缓存，下面给出中间结果定义。
定义4 在RTMR方法中，将MapReduce模型中的［k2，List〈v2〉］和list（v2）称为中间结果。
借鉴 Metis［7］的思路，中间结果在内存中采用Hash B＋树存储（如图3），具有相同Hash值的k2使用B＋树存在同一 Hash表项中，［k2，List（v2）］使
用线性表存储并链接在B＋树的叶节上，list（v2）存储在叶节点。Hash B＋树结构具有很高的读写性能，如果k2具有唯一的Hash值，则可以通过分配足够的项来避免Hash冲突，插入和查找操作的复杂度仅为O（1）；如果k2没有唯一的Hash值，插入和查找的复杂度仅为也只有O（1）＋O（log n）。为了扩大中间结果的本地存储容量，在外存上构造包括一个索引块和多个64KB数据块的SSTable文件［8］（如图4），以块为单位为Hash表项分配外存空间。在数据流处理过程中，如果所需的中间结果表项不在内存且内存已无空间时，则发生内外存替换。
为了提高数据流处理能力，RTMR在Map和Reduce阶段间构造阶段流水线（如图5），每个阶段由线程池、输入缓冲区和阶段控制器组成，利用线程池减少每次处理时的初始化开销，并通过在缓冲区中异步传递数据消除阶段间的同步。在本地化架构下，阶段流水线可以通过阶段内的批调整和阶段间线程池控制调整资源分配，优化对CPU的利用。
3 RTMR集群适应性构造技术
从实现角度，RTMR集群架构由 Map和Reduce节点的配置和拓扑结构决定。在RTMR中，为充分利用节点计算能力，Map节点同时也充当Reduce节点，将配置x个Map节点的架构称为RTMR（x）。例如，在4节点的集群中，图6a～图6c所示分别为1，2和4个 Map节点的架构RTMR（1），RTMR（2）和 RTMR（4）。针对不同的应用特征、节点计算能力和网络条件，如何适应性地构造最优架构是一个关键问题。在数据流处理系统中，衡量处理能力的指标是数据流平均处理延迟，选择最优架构的目标是使处理延迟最小。
针对RTMR应用，若集群有n个节点，每个节点在一定历史数据规模下能够达到的Map和Reduce阶段处理速度分别为μm和μr，Map节点的个数为x，在数据流速度为λm的情况下，Map阶段相当于一个M／M／x排队系统。根据M／M／x排队论模型［9］，经过Map阶段的数据流平均延迟为 式中
在Reduce阶段数据流速度为λr的情况下，因为所有节点都是Reduce节点，所以Reduce阶段相当于一个M／M／n排队系统。根据排队论模型，经过Reduce阶段的数据流平均延迟为 式中
集群采用交换机连接，若两节点间的网络传输速度为μn，RTMR（x）架构中Map节点的输出连接有n－1个，Reduce节点的输入连接有x个，每个连接的传输速度与总连接数成反比，即
每个连接上的数据流速度为，根据 M／M／1排队模型［12］，经过一个 Map节点的数据流网络延迟为
并行经过所有Map节点的数据流平均网络延迟
在集群中，节点上的线程增加将引起额外开销。对应于网络连接，每对Map与Reduce节点间存在n＋x－1个线程用于接收和发送数据，若延迟因子为ε，则经过一个Map节点的数据流额外延迟
并行经过所有Map节点的数据流平均额外延迟
综上，RTMR（x）的数据流处理延迟为
显然，L（x）是 Map节点数x的单调减函数。因此，对于n个节点的集群来说，RTMR（n）的并行处理程度最高，平均延迟最小。
此外，由定理2可知还存在一类如图6d所示的架构：将预处理的中间结果分布缓存于各节点，每个工作节点冗余接收数据流，通过Map阶段过滤出本节点负责处理的数据，并在本地缓存上进行Reduce计算。将这种架构定义为RTMR（0），RTMR（0）避免了网络传输和额外延迟，经过每个节点的数据流相当于经过一个M／M／1的Map阶段和一个M／M／1的Reduce阶段，处理延迟
并行经过n个节点的数据流处理延迟
适应性选择最优架构就是比较L（n）和L（0）。 4 RTMR集群负载均衡算法
从集群角度来说，负载倾斜将使系统整体吞吐量受制于过载节点，因此如何实现负载均衡是保障数据流处理可伸缩性的关键问题。造成集群负载倾斜的原因一方面是节点上的历史数据分布不均，另一方面是到达各节点上处理的数据流分布不均。针对上述两种情况，需要从静态和动态两方面考虑，在初始化时根据节点的处理能力划分历史数据，在运行时动态地调整节点的历史数据。 4．1 问题定义
在到达数据流均匀分布的情况下，根据各节点的处理能力划分历史数据能够平衡各
节点的处理负载，消除单点瓶颈，提升整体吞吐量。数据划分的关键是建立刻画节点处理能力的性能模型，若节点内存容量为M，历史数据划分区间为P，命中率提升因子为α，则中间结果内存命中率η＝αM／P，非命中率。若节点在内存命中情况下的处理速度为μ，非命中情况下的阻尼因子为ξ，则平均处理速度
在负载均衡情况下，各节点的处理速度相同，因此节点所分配的历史数据规模
基于上述模型，可以根据CPU速度和内存容量计算出各节点所分配的历史数据规模，在数据流均匀分布的情况下，能够保证各节点的负载均衡。
实际运行中，由于到达的数据流负载为非均匀分布，在经过长时间运行之后，可能造成热点数据，从而导致节点间的负载倾斜。因为无法控制外部数据流，所以为了消除负载倾斜，只能通过调整数据分布的方法平衡各节点上的实际处理负载。动态调整数据分布的关键是找出过载节点和空闲节点并确定调整哪部分数据，在实现负载均衡的同时最小化移动数据的代价。
将数据流看作时间序列，基于预测方法（如灰色预测、指数平滑等）建立节点的状态模型。在下一调整周期，若表项i上的数据流负载为Δi，节点j的历史数据区间为Pj，则节点的数据流负载为
定义5 利用式（3）计算节点j能够处理的负载pj。设过载阈值为φ，若sj＞ （1＋φ）pj，则称节点j过载；设空闲阈值为τ，若sj＜（1－τ）pj，则称节点j空闲。 针对过载节点集合ON，每个过载节点j的历史数据区间为Pj，则可能发生数据移动的区间
针对空闲节点集合IN，设xik表示表项i到节点k的移动，xik＝1时表示移动，
xik＝0时表示不移动。若节点k表项i的数据量为dik，则从过载节点到空闲节点的数据移动量
并且每个表项只能移动一次，即
为达到负载均衡，每个过载节点要移动数据达到非过载状态，即
每个空闲节点要接收数据达到非空闲状态，即
每个空闲节点不能接受过多数据成为过载节点，即
综上所述，负载均衡动态调整可建模为在约束（5）～（8）下最小化式（4）的0／1规划问题。
动态负载均衡问题是NP难的［10］。在求解算法中，执行效率和数据移动代价影响对负载倾斜的反应速度，均衡程度决定吞吐量提升效果，算法设计要权衡这三个因素。为了度量节点的负载倾斜程度，定义负载倾斜度。 定义6 负载倾斜度
其中对于过载节点和空闲节点，负载倾斜度分别指过载程度和空闲程度。 为了度量算法的负载均衡效果，定义负载均衡度。 定义7 负载均衡度
在动态负载均衡中，应对负载倾斜的反应速度由算法执行时间和数据移动代价两方面决定，为了度量算法的反应速度，给出如下定义：
定义8 动态负载均衡算法的反应速度为执行时间与数据移动代价之和，即
式中：l为迭代次数，d为传输数据量，Coste（）和Costm（）分别为执行时间和传输时间函数。
为了综合衡量算法的执行效率、数据移动代价和负载均衡效果，定义均衡效率如下： 定义9 均衡效率为负载均衡度与反应速度的比值，即 4．2 分解算法
借鉴子问题分解的思路，首先对集合ON中所有过载节点按照过载程度由大到小排序，对集合IN中所有空闲节点按照空闲程度由大到小排序，之后通过计算节点的负载状态转换关系对空闲节点和过载节点进行分组，每个分组形成一个子问题的输入（如图7算法1流程图）。在算法1中，每次分组时建立过载节点集合ON′和空闲集合IN′，加入ON和IN 中未分组的第一个节点，形成分组［ON′，IN′］。计算IN′中所有空闲节点达到非空闲状态所要增加的负载量
达到过载状态所要增加的负载量
计算ON′中所有节点达到非过载状态所要减少的负载量
如果D3≤D1，表明空闲节点在接受过载节点到达非过载状态减少负载后还不会到达非空闲状态，则ON′加入ON 中下一未分组的过载节点；如果D2≤D3，表明空闲节点无法承受过载节点到达非过载状态所减少的负载，则IN′加入IN 中下一未分组的空闲节点；如果D2＞D3＞D1，则一次分组完成。重复上述过程，直到过载或空闲集合为空。若空闲集合为空，则表示集群无法应对当前负载，需要额外增加节点。
与仅考虑两两节点或一个过载节点与多个空闲节点间平衡的方法不同，算法1通过考虑过载节点和空闲节点间的负载状态转换关系，能够针对各种负载倾斜进行分组，并在节点数为n的情况下，复杂度仅为O（n）。 4．3 搜索算法
针对每个子问题，在过载节点数据表项规模扩大的情况下解空间仍然很大，因此应采用模拟退火算法或遗传算法等启发式算法［10－11］。然而，上述算法需要设定最大迭代次数，迭代次数太少则解的准确性较差，迭代次数太多又会增加算法的执行时间。实际上，由定义8知，应对负载倾斜的反应速度由算法执行时间和数据移动代价两方面决定，因此5．1节问题定义中的优化目标应调整为minν。 针对此优化目标，在寻求最优移动代价解的同时要避免不必要的迭代，应综合已执行时间和当前解作为判定标准。设已迭代次数为l，当前解的数据量为dl，当前最优反应速度为vopt，若Coste（l）＋Costm（dl）≤vopt，则表示算法执行到当前，找到反应速度更优的解。在找到当前最优解vopt后，若重复迭代Lm 次，对于任意i∈［1．．Lm］，都有Coste（l）＋Costm（dl）≤Coste（l＋i）＋Costm（dl＋i），则认为已找到反应速度近似最优的解并返回。同时，为避免陷入 局 部 最 优 解，若 存 在i ∈ ［1．．Lm］，使 得Costm（dl）＞Costm（dl＋i），则接受这个解，重新开始上述过程。
算法2根据上述原理改进了模拟退火算法（如图8），以贪心方式初始化解，然后进入模拟退火迭代过程，若找到比当前最优反应速度更优的解，则接受这个解，清空重复迭代次数计数器继续迭代；否则，进入当前最优反应速度的重复迭代，当重复迭代次数已满时，以当前最优解作为近似最优解返回。若在重复迭代过程中找到比当前最优的数据移动代价差的解，则以一定概率接受此解，以避免算法陷入局部最优解。同时，为了保证算法能够快速收敛，在接受此较差解后将重复迭代次数减半。
算法2中最关键的是确定重复迭代次数，设负载倾斜涉及的表项数量为h，则动态负载均衡问题规模s＝2h。贪心算法的复杂度Lg＝h2log h，不失一般性，设模拟退火算法和算法2的最大迭代次数，若重复迭代次数为Lm，则在最坏情况下，算法2当前最优解的重复迭代次数为
由RTMR过程可知，表项数量由二进制Hash函数决定，问题规模按指数方式增长，即h＝2k。可设重复迭代Lm＝alog h＋b，这样在最坏情况下的重复迭代仅为O（k），能够保证算法2在有机会得到更优反应速度的同时及时收敛，从而兼顾执行效率和数据移动代价，优化反应速度。 5 评价
本章以城市车辆监管系统中具有代表性的套牌车计算作为基准测试来验证本文方法。 5．1 基准测试
套牌计算根据车辆时空矛盾来判定，针对每一条出现在特定地点的车辆实时数据，检索该车牌出现在其他点且在最大套牌时间阈值内的历史数据，如果二者的时间差小于两点间的套牌时间阈值，则认为该车辆有套牌嫌疑。在某大型城市，车牌数量达到108级别，若全面捕获车辆数据，高峰时将以10MB／s的速度产生数据流（每条数据按500Byte计，约20 000条／s），同时在处理窗口为1d的情况下，历史数据规模将达到1TB。套牌计算可以使用如下RTMR算法：针对某个车牌号，Map方法在所有车牌号的Hash表中找到其所在表项，Reduce方法在B＋树中找到其链表所在位置，依次与每条历史数据比较套牌时间阈值并更新链表。算法可以采用输出为20位二进制数的Hash函数hash（k）＝k mod 220，存储中间结果的Hash表共有220个表项，平均每个Hash表项存储108／220≈100个车牌的数据。
在确定计算过程后，还要确定基准测试的负载构造方法。若数据流负载在算法1后的分组为［ON′ ，IN′ ］，由5．1节可知D1 ，D2 和D3 的计算方法，根据IN′和ON′节点数不同可以分为三种负载。
定义10 若对于负载的任意分组［ON′，IN′］，都有｜ON′｜＝1且｜IN′｜＝1，则此负载为1－1型。
构造1－1型负载就是任意选择节点对［i，o］，确定各节点的负载量，使得D1＜D3＜D2。
定义11 若对于负载的唯一分组［ON′，IN′］，有｜ON′｜＝1且｜IN′｜＝n（n≥1），则此负载为1－n型。
构造1－n型负载就是选择一个过载节点和任意空闲节点的分组［o，IN′］，确定各节点的负载，使得D1＜D3＜D2
定义12 若负载的任意分组［ON′，IN′］，都有｜ON′｜＝n1（n1≥1），｜IN′｜＝n2（n2≥1），则此负载为n－n型。
构造n－n型负载就是选择任意数量过载节点和空闲节点分组，确定各节点的负载量，使得D1＜D3＜D2。
根据车辆数据流的随机性和局部性特点，数据流的模拟方法为：以十进制区间［1，108］内的数模拟车牌；将n个节点的历史数据划分为P1，P2，…，Pn，在各节点的历史数据划分区间中选取一个子集P1′，P2′，…，Pn′，使得｜P1′｜＋｜P2′｜＋…＋｜Pn′｜＝105，均匀负载是循环地为n个节点产生负载，非均匀负载则按照各种类型构造方法选择节点产生负载；对于节点i，在Pi′中选取一个随机目标表项t，在区间［1，100］内选取一个随机数x，以220 x＋t作为该条模拟数据的车牌号，随机设定监控点，记录系统时间添加时间戳并控制数据流速度。 实验环境方面，RTMR系统搭建在2×4核2．0 GHz CPU、32GB内存和250GB硬盘服务器集群上，使用一台4×4核2．4GHz CPU、64GB内存服务器作为控制
节点；使用Oracle 10g作为持久化存储，搭建在一台2×4核2．4GHz CPU、16GB内存服务器和20TB RAID5磁盘阵列上；网络连接采用1Gbps以太网光纤和交换机；在一台双核3．0GHz CPU和4GB内存服务器上使用Load Runner 9．0模拟数据流。 5．2 适应性架构分析
实验1在4节点集群环境下对比RTMR（0），RTMR（1），RTMR（2）和 RTMR（4）在不同历史数据规模下的数据流处理能力。每种数据规模测试10次，每次10min，取平均值计算实验结果。由图9可知，无论在何种数据规模下，RTMR（4）、RTMR（2）和RTMR（1）的数据流处理能力均依次降低，即随着Map节点的增加，并行程度增强，数据流处理能力提高。此外，在数据流速度高于15MB／s时，RTMR（0）的处理能力低于 RTMR（1），RTMR（2）和RTMR（4），这是因为RTMR（0）采用广播模式，在数据流速度较高的情况下，每个节点接收数据和进行Map处理占用的CPU时间较多，影响了用于Reduce阶段的CPU时间，从而制约了整体处理能力。随着数据规模增加到200GB，各种架构的数据流处理能力都下降到15MB／s以下，此时RTMR（0）的处理能力开始高于 RTMR（1），RTMR（2）和RTMR（4），这是因为每个节点接收数据和进行Map处理的开销已经不影响Reduce阶段，同时也避免了数据传输等额外开销。
5．3 可伸缩性分析
本节采用伸缩性分析比较RTMR（0）和RTMR（n）针对历史数据规模和数据流速度的伸缩能力。实验2中固定数据流速度为1MB／s，测试在节点增加情况下所能处理的历史数据规模。由图10可知，RTMR（0）在节点增加时处理能力的提升是近似线性的，这是由于RTMR（0）通过划分历史数据中间结果和本地化处理，
使节点之间不会产生制约并行吞吐量提升的数据传输和同步开销，之所以未能达到线性伸缩，是因为在数据规模扩大情况下增加了本地存储文件的读写开销；而RTMR（n）随着节点增加，节点之间的数据收发和传输开销增加，限制了所能处理的数据规模。实验3中固定每个节点数据规模为50GB，测试在节点增加情况下所能处理的数据流。由图11可知，随着节点的增加，RTMR（n）的数据收发和传输开销虽然增加，但针对数据流速度的伸缩能力要强于RTMR（0）。这是因为RTMR（n）将数据流分布到各个节点上并行处理，而RTMR（0）在数据流速度提高时每个节点接收数据和进行Map处理的CPU开销增加，制约了Reduce阶段的处理能力。具体来说，RTMR（0）在数据流速度低于15MB／s时，处理能力的提升是近似线性的，当数据流速度超过15MB／s时，处理能力提升变缓。
在实际应用中，利用模型分析技术，可以根据应用特征和集群环境适应性确定最优架构。从利用RTMR方法解决车辆监管应用的经验来看，当前物联网环境下的数据流处理应用受采集端带宽等因素的影响，数据流速度远达不到15MB／s，只需占用服务器多核CPU的很小一部分就可以完成接收和Map操作，在历史数据规模不断扩大的情况下，RTMR（0）更具适用性。 5．4 负载均衡分析
实验4是在RTMR（0）架构下验证RTMR负载均衡方法的效果。该实验搭建在4节点集群上，节点1，2和3使用2×6核2．0GHz CPU、32GB内存服务器，节点4使用1台2×4核800MHz CPU、8GB内存服务器，模拟两种数据流负载：①在4个节点上均匀分布的负载；②在节点1和节点2上分布90%数据的n－n型非均匀负载测试1min内的吞吐量变化，非均匀负载从第10s开始，实验进行10次，取平均值计算实验结果，动态负载均衡算法的运行周期固定为500ms。
由图12可以看出，在均匀负载下，RTMR划分方法由于根据节点能力差异进行分配，避免了单点瓶颈，与均匀划分方法相比，其整体吞吐量更高；在非均匀负载下，若只进行静态数据划分，则在第10s开始出现负载倾斜，吞吐量迅速下降。使用动态调整方法能够及时发现负载倾斜，定位过载节点和空闲节点并进行数据移动，在第20s时重新达到负载均衡。 5．5 分解算法性能
为降低动态负载均衡问题的复杂度，可以通过子问题分解简化问题规模，文献［12－13］提出考虑两两节点之间平衡的1－1类型分解法，文献［14］提出考虑一个过载节点与多个空闲节点平衡的1－n类型分解法。实验5在16节点集群（每节点216个表项）环境下，比较分枝限界算法以及先分解子问题再使用分支限界的算法在各种负载类型下的均衡效率，实验结果以分支限界法的均衡效率为基准做归一化处理。由图13可知，无论何种负载，单纯使用分枝限界算法虽然能够取得很好的均衡效果，但是执行效率太低，因此均衡效率最差。对于1－1型和1－n型负载，分别使用1－1法和1－n法分解子问题能够得到较高的均衡效率，RTMR分解法的均衡效率与其非常接近。然而，在n－n型负载下，使用1－1分解法和1－n分解法的执行效率虽然较高，但均衡效果太差，造成均衡效率急剧降低，而RTMR能够在较短时间内达到较好的均衡效果，均衡效率依然较高。
实验6分析子问题分解算法在不同节点数目下的均衡效率。实验按照1－1负载和n－n负载，以及1－1分解法和RTMR分解法分为四组。由图14可知，在节点数目增加的情况下，无论对于何种负载，因为RTMR分解算法只有O（n）的复杂度，所以均衡效率衰减得非常缓慢，特别是对于n－n型负载，1－1分解法很快失效，而RTMR分解法依然具有较高均衡效率。1－n型负载和分解法可以得到
类似结论。 5．6 搜索算法性能
针对NP难的动态负载均衡问题，文献［10，13］分别提出了基于贪心方式和基于模拟退火方式的搜索算法。实验7在不同问题规模情况下比较了贪心算法、模拟退火算法和RTMR搜索算法的动态负载均衡反应速度。实验中，负载均衡涉及的数据项呈指数增长，h＝2k，设RTMR算法重复迭代次数Lm＝2k，各种算法参数设置与5．3节相同。由图15可知，随着问题规模的扩大，模拟退火算法虽然能得到较优的数据移动代价，但因为需要大量迭代影响了执行效率；贪心算法虽然执行效率较高，但是在问题规模不断扩大的情况下，不能保证得到较优的数据移动代价；RTMR算法与遗传算法相比，大大减少了迭代次数，与贪心算法相比，在问题规模扩大的情况下，重复迭代保证算法有机会找到更优解并且及时收敛，在寻找最优数据移动代价的同时兼顾执行效率，因此能够获得最佳的反应速度。 5．7 相关工作比较
针对MapReduce的实时性改进已经成为一个研究热点。增量处理 Percolator［15］和迭代处理Twister［16］等工作通过随机访问存储和中间结果缓存改进了大规模数据处理的性能，但仍属于对静态数据增量的批处理，不适合针对高速数据流进行大规模数据处理；HOP［17］和 S4［18］利用流水线技术扩展了MapReduce的数据处理能力，但依然不是以大规模历史数据基础上的流处理为目标，这表现在缺乏对历史数据进行预处理和缓存的支持，并且需要程序员依靠经验方式配置节点类型和拓扑结构，缺乏适应性的架构构造机制。本文通过中间结果缓存、本地化和流水线技术扩展了MapReudce的数据流处理能力，并通过模型分析适应性的构造集群架构。
针对动态负载均衡问题，使用分枝限界等精确性算法虽然能够得到最优解，达到很好的负载均衡效果，但随着问题规模的增大将发生组合爆炸，执行效率制约了应对负载倾斜的反应速度，仅适用于问题规模较小的情况。在问题规模扩大的情况下，文献［10－11］使用模拟退火、遗传算法等启发式算法寻找近似解，但启发式算法需要设定最大迭代次数，迭代太少则解的准确性较差，迭代太大则会影响算法的执行效率。还有方法通过子问题分解简化问题规模［12－14］，文献［14］在数据库集群中通过权值累加计算节点处理能力，并将过载节点的多余数据平均分配到空闲节点，仅适用于一个过载节点多个空闲节点的情况；文献［12］提出的Flux算子能够动态平衡集群上的历史数据，但其算法在子问题分解时仅涉及了两两节点之间平衡，不能达到全局平衡。本文根据节点负载状态转换关系提出了一种O（n）复杂度的子问题分解算法，适用于各种负载倾斜情况，并针对子问题提出了一种综合执行时间和数据移动代价为优化目标的搜索算法，提高了应对负载倾斜的反应速度。 6 结束语
针对面向大规模历史数据的数据流处理可伸缩性问题，本文提出了一种RTMR方法，其创新性表现在：
（1）通过中间结果缓存、本地化和流水线技术，改进了MapReudce的数据流实时处理能力。
（2）基于模型分析提出了一种适应性的RMT集群构造技术，该技术能够根据应用特点和集群环境配置MapReduce节点类型、数量和拓扑结构。
（3）在根据节点处理能力划分历史数据的基础上，提出了一种动态负载均衡算法，通过计算节点的负载状态转换关系将问题分解为若干子问题，然后综合考虑算法执行时间和数据移动代价求解各个子问题，在保障集群全局平衡的同时寻找反应速度近似最优的调整方案。
在RTMR方法中，编程模型、实现架构和负载均衡的是集群可伸缩性的基础，除此之外，单个节点处理能力的提升也是保证集群可伸缩性的关键。因此，下一步工作主要包括RTMR中间结果读写性能优化和阶段流水线性能优化。
【相关文献】
［1］MOTWANI R，WIDOM J，ARASU A，et al．Query processing，resource management，and approximation in a data stream management system［C］／／Proceedings of the 1st Biennial Conference on Innovative Data Systems Research．New York，N．Y．，USA：ACM Press，2003：176－187．
［2］ABADI D J，AHMAD Y，BALAZINSKA M，et al．The design of the Borealis stream processing engine［C］／／Proceedings of the 2nd Biennial Conference on Innovative Data Systems Research．New York，N．Y．，USA：ACM Press，2005：277－289． ［3］JIN Cheqing，QIAN Weining，ZHOU Aoying．Analysis and management of streaming data：a survey［J］．Journal of Software，2004，15（8）：1172－1181（in Chinese）．［金澈清，钱卫宁，周傲英．流数据分析与管理综述［J］．软件学报，2004，15（8）：1172－1181．］
［4］DEAN J，GHEMAWAT S．MapReduce：simplified data processing on large clusters［J］．ACM Communication，2008，51（1）：107－113．
［5］SHAH M A，HELLERSTEIN J M，CHANDRASEKARAN S，et al．Apache hadoop［EB／OL］．［2011－08－17］．http：／／hadoop．apache．org／．
［6］RANGER C，RAGHURAMAN R，PENMETSA A，et al．E－valuating map reduce for multi－core and multiprocessor systems［C］／／Proceedings of the 13th International Conference on High－Performance Computer Architecture．Washington，D．C．，USA：IEEE Computer Society，2007：13－24．
［7］KAASHOEK F，MORRIS R，MAO Y．Optimizing MapReduce for multicore architectures［R］．Boston，Mass．，USA：MIT Computer Science and Artificial Intelligence Laboratory，2010．
［8］CHANG F，DEAN J，GHEMAWAT S，et al．Bigtable：a distributed storage system for structured data［C］／／Proceed－ings of the 7th Symposium on Operating Systems Design and Implementation．Berkeley，Cal．，USA：USENIX Association，2006：205－218．
［9］DIAO Zaijun，ZHENG Handing，LIU Jiazhuang，et al．Operational research
［M］．Beijing：Higher Education Press，2001：242－255（in Chinese）．［刁在筠，郑汉鼎，刘家壮，等．运筹学［M］．北京：高等教育出版社，2001：242－255．］ ［10］HEISS H，SCHMITZ M．Decentralized dynamic loadbalancing：the particles approach［J］．Information Sciences，1995，84（2）：115－128．
［11］LIU Zhenying，FANG Binxing，HU Mingzeng，et al．An effective load balance method［J］．Journal of Software，2001，12（4）：563－569（in Chinese）．［刘振英，方滨兴，胡铭曾，等．一个有效的动态负载平衡方法［J］．软件学报，2001，12（4）：563－569．］
［12］SHAH M A，HELLERSTEIN J M，CHANDRASEKA R S，et al．Flux：an adaptive partitioning operator for continuous query systems［C］／／Proceedings of the 19th International Conference on Data Engineering．Washington，D．C．，USA：IEEE Computer Society，2003：25－36．
［13］DENG Huafeng，LIU Yunsheng，XIAO Yingyuan．Dynamic
load balancing techniques for the distributed stream processing system［J］．Computer Science，2007，34（7）：120－123（in Chinese）．［邓华锋，刘云生，肖迎元．分布式数据流处理系统的动态 负 载 平 衡 技 术 ［J］．计 算 机 科 学，2007，34（7）：120－123．］ ［14］GONG Weihua．Study on the key issues of database cluster system［D］．Wuhan：Huazhong University of Science and Technology，2006（in Chinese）．［龚卫华．数据库集群系统的关键技术研究［D］．武汉：华中科技大学，2006．］
［15］PENG D，DABEK F．Large－scale incremental processing using distributed transactions and notifications［C］／／Proceedings of the 9th USENIX Symposium on Operating Systems Design and Implementation．Berkeley，Cal．，USA：USENIX Association，2010：251－264．
［16］EKANAYAKE J，LI H，ZHANG B，et al．Twister：a runtime for iterative MapReduce［C］／／Proceedings of the 19th ACM International Symposium on High Performance Distributed Computing．New York，N．Y．，USA：ACM Press，2010：810－818． ［17］CONDIE T，CONWAY N，ALVARO P，et al．MapReduce online［C］／／Proceedings of the 7th USENIX Symposium on Networked Systems Design and Implementation．Berkeley，Cal．，USA：USENIX Association，2010：313－328． ［18］NEUMEYER L，ROBBINS L，NAIR A，et al．S4：Distributed stream computing platform［C］／／Proceedings of the 10th IEEE International Conference on Data Mining Workshops．Washington D．C．，USA：IEEE Computer Society，2010：170－177．
因篇幅问题不能全部显示，请点此查看更多更全内容