基于Storm的海量数据实时聚类.pdf

《基于Storm的海量数据实时聚类.pdf》由会员分享，可在线阅读，更多相关《基于Storm的海量数据实时聚类.pdf（8页珍藏版）》请在淘文阁 - 分享文档赚钱的网站上搜索。

1、龙源期刊网 http:/基于基于 StormStorm 的海量数据实时聚类的海量数据实时聚类作者：王铭坤 袁少光 朱永利 王德文来源：计算机应用2014 年第 11 期摘 要：针对现有平台处理海量数据实时响应能力普遍较差的问题，引入Storm分布式实时计算平台进行大规模数据的聚类分析，设计了基于Storm框架的 DBSCAN算法。该算法将整个过程分为数据接入、聚类分析、结果输出等阶段，在框架预定义的组件中分别编程实现，各组件通过数据流连通形成任务实体，提交到集群运行完成。通过对比分析和性能监测，验证了所提方案具有低延迟和高吞吐量的优势，集群运行状况良好，负载均衡。实验结果表明Storm平台处理

2、海量数据实时性较高，能够胜任大数据背景下的数据挖掘任务。关键词：Storm；海量数据； 聚类；实时分析中图分类号： TP316.4；TP311.1 文献标志码：A 0 引言聚类分析是数据挖掘中常用的分析手段，由于聚类算法的复杂度一般比较高1，利用现有单机环境进行海量数据的聚类分析存在吞吐量小、实时性差、难以达到预期效果的缺陷2，已难以满足大数据背景下的业务需求，通过集群处理的分布式计算模型为此类问题的解决提供了新途径。 Storm是开源的分布式实时计算框架，可以高效地处理大量数据流3。本文拟将 Storm平台应用到海量数据的聚类分析，设计Storm框架下的 DBSCAN（DensityBase

3、d SpatialClustering of Application with Noise）算法，搭建高性能的集群环境，对海量数据进行实时高效的分析与处理。 1 相关研究 1.1 大数据处理研究现状大数据是信息技术产业又一次颠覆性的技术变革，并已在多个领域证明了它的重要性和价值，许多机构和公司都在利用大数据技术获取价值。大数据研究的不断深入也带动了相关技术的发展，面向不同应用场景的大数据处理平台相继出现。大数据的处理可以简单分为批处理与实时流处理两种模式。 Hadoop是批处理平台的代表，它主要由分布式文件系统（Hadoop Distributed FileSystem， HDFS）、MapR

4、educe编程模型和 Habase分布式数据库组成4。用户可以在不了解分布式底层细节的情况下开发分布式程序，其强大的吞吐能力、高容错性以及良好的可扩展性龙源期刊网 http:/保证了大规模数据的可靠处理4。但是此类批处理系统的固有缺点是实时响应能力差，数据处理延迟较长，无法满足用户的实时性要求高的需求。针对 Hadoop的上述不足，新一代大数据分析引擎Impala应运而生，它提供结构化查询语义，采用查询优化技术实现HDFS和 HBase中数据的快速查询5。Impala适合用来处理输出数据适中或比较小的查询，作为MapReduce的一个强力补充，未能从根本上提升批处理系统的实时处理能力。另外，S

5、park是一个立足于内存计算的分布式计算系统，它从多迭代批量处理出发，支持流处理和图计算等多种范式6。Spark对小数据集的处理能达到亚秒级的延迟，处理大数据集时相对于 MapReduce也有较大提升，这主要得益于Spark的内存计算、数据本地性和传输优化、调度优化等技术。但是Spark不能很好地支持细粒度、异步的数据处理，并且在性能、稳定性和可扩展性上还有较大的提升空间。综上所述，上述批处理系统及其衍生系统都未能从根本上提升大规模数据的实时响应能力， 2011年 Twitter对 Storm的开源使这个情况大为改观3。Storm开源后在淘宝、百度等各大互联网公司得到了充分应用3，比如淘宝的虫

6、洞系统以Storm实时流处理为引擎，进行实时流量分析、个性化推荐等。第 11期 王铭坤等：基于 Storm的海量数据实时聚类计算机应用 第 34卷 1.2 Storm平台 Storm是一个分布式实时计算系统，与Hadoop等批处理系统不同，Storm专注于流数据的处理，主要应用于实时分析、持续计算、在线机器学习、分布式远程调用以及数据抽取、转换和加载等领域7。在 Storm集群中有两种节点：控制节点和工作节点。控制节点上面运行着名为Nimbus的后台程序，负责在集群里面分发代码，分配计算任务给机器，并且监控集群状态。每个工作节点上面运行着名为 Supervisor的后台程序，Superviso

7、r会监听分配给它的机器的工作，根据需要启动或关闭工作进程。在 Storm框架中，计算任务的逻辑被封装到Topology对象里面。Topology是由不同的Spout和 Bolt通过数据流（Stream）连接起来的图，如图 1所示。Spout是 Topology的消息生产者，通常会从外部数据源（消息队列、数据文件、网络传输等）读取数据，然后以Tuple的形式发送到 Stream中。Bolt中封装的是处理逻辑，接收到Tuple后可以执行过滤、聚合、计算、函数操作等。Spout和 Bolt各组件之间是订阅关系，可以灵活实现数据的定向与分流。 2 基于 Storm的 DBSCAN算法 2.1 DBSC

8、AN算法龙源期刊网 http:/ DBSCAN是一种基于密度的聚类算法8，它把密度足够高的区域划分为簇，并能够在具有噪声的数据中发现任意形状的簇9。相对于其他聚类算法，DBSCAN具有可以处理高维数据和噪声对象排除效果好的优点10-11。首先说明 DBSCAN算法的几个重要定义12：定义 1 密度。空间中某一点的密度是以该点为圆心、以Eps为半径的圆内包含点的数目。定义 2 邻域。空间中某一点的邻域是以该点为圆心、以 Eps 为半径的圆内包含点的集合。定义 3 核心点。空间中某一点的邻域内若至少包含最少MinPts个点，则该点为核心点。定义 4 边界点。空间中某一点的密度如果小于阈值MinPt

9、s，则称该点为边界点。定义 5 直接密度可达。在给定的集合D 中，如果点 p 在 q的邻域内，并且 q是核心点，那么称点 p从点 q是直接密度可达的。定义 6 密度可达。如果存在集合p1， p2， pn， p1=q， pn=p， 对 piD（1in），pi+1是从 pi 直接密度可达，则点 p是从点 q 密度可达的。定义 7 密度相连。在给定的集合 D 中，如果 p和 q都是从 o 密度可达的，那么称p 和 q是密度相连的。定义 8 类与噪声。集合 D 的非空子集 C 是一个类，当且仅当C满足以下条件： 1）对于任意的 p，qD，如果 qC，同时 p是从 q 密度可达的，那么pC； 2）对于任

10、意的 p，qC， p和 q 是密度相连的。不属于任何类的对象被认为是噪声。 DBSCAN 算法的具体步骤12如下：第 1步 计算各点的密度，找出核心点，点之间的距离通常用欧氏距离公式计算。第 2步 逐个扫描点，判断是否已经聚类：如果是，则跳过；如未聚类则执行第3步。第 3步 判断该点是否为核心点，如果是核心点，找出所有从该点密度可达的对象，形成一个新的类，对它的下一个点进行判断，如果在它的邻域内，则设置为相同的类。第 4步 下一个点还是核心点，则递归进行处理，直到所有点处理完。龙源期刊网 http:/ DBSCAN对用户定义的参数 Eps和 MinPts很敏感，参数的选择无规律可循，一般靠经验

11、确定11，13。 2.2 Storm框架下 DBSCAN算法实现在 Linux集群上搭建 Storm完全分布式环境，首先部署Storm运行依赖的 JDK、Zookeeper和 Python等外部资源，再部署 Storm。图 2是 Storm集群架构与执行流程。 2.2.1 数据接入模块本文的目的是利用 Storm分布式实时计算框架进行海量数据的聚类分析，核心思想是将计算任务分配给多个节点，最后汇总处理结果，通过任务分配与调度达到提升计算性能的目的。在 Storm中，Spout组件是整个框架的数据源头，用户通过实现IRichSpout接口接入数据。读入原始数据后，首先要按照算法设计进行分割，每1

12、0000个数据封装为一个 Tuple元组，然后发送到待处理队列。为了保证数据的顺序性，发送的每个Tuple会有唯一 Id，Id中保存了 Tuple在队列中的具体位置，最后会根据Id的顺序来汇总计算结果。数据读入的流程如图3 所示，由于 Storm中的进程以服务的形式存在，如非人为杀死，这些进程将会一直运行，所以该流程是没有结束标记的。此外，Spout的 ack方法和 fail方法在元组被成功处理和处理失败时分别调用，保证了数据处理的完整性。 2.2.2 聚类分析模块在 Storm框架中，逻辑运算的任务主要在Bolt环节实现，Bolt各组件与 Spout是灵活的订阅关系，形成了松耦合的管道过滤器

13、体系结构，这一设计极大提升了Storm的可扩展性。根据系统设计，聚类过程分为两个阶段，首先对各个Tuple中的数据进行聚类，按照原始DBSCAN算法在名为 SingleDbscan的 Bolt中完成，单次聚类结果将被发送到名为AllDbscan的 Bolt中进行汇总。汇总过程中，首先找出单次聚类产生的核心点，然后对这些核心点进行DBSCAN聚类。按照 DBSCAN算法的思想，如果两个簇的核心点是一类，那么这两个核心点是互相直接密度可达的，它们代表的簇要归为一类。通过遍历筛选后，最终不属于任何类的点则为噪声对象。聚类分析的相关逻辑SingleDbscan和 AllDbscan封装在不同的 Bol

14、t中。Bolt过程主要包括准备、执行和清理三个阶段，分别对应预定义的prepare、 execute和 cleanup函数，开发人员在各函数中编程实现即可。在prepare函数中声明执行对象和输出对象，获取相关资源；execute函数是计算任务执行的主体； cleanup阶段主要是释放和清理资源，保证后续任务顺利执行。龙源期刊网 http:/ Storm框架本身不负责计算结果的保存，可以实现不同的Bolt来完成计算结果的多样化存储。本文对 DBSCAN聚类结果的处理方式有DatabaseBolt和 FileBolt。DatabaseBolt将聚类结果保存到数据库，便于查询检索； FileBol

15、t将聚类结果直接存储到数据文件中，便于复制和迁移。 2.2.3 主函数主函数的主要任务是设置作业执行相关参数，主要有以下步骤： 1）实例化 TopologyBuilder对象，TopologyBuilder的作用是创建任务执行的实体。 2）通过 Config对象设置端口、路径等配置参数，任务开始后，具体配置信息将分发给不同的机器。 3）调用 SetSpout函数实现数据源的接入，设置其任务执行数量，声明输出字段，后续的Bolt将按照可选的流聚合方式订阅Spout的输出数据。 4）调用 SetBolt函数，根据业务需求设定Topology中的任务组成。合理设置各Bolt的处理顺序和并行任务数量，

16、使更多的计算资源分配给性能瓶颈，以此提升程序的运行效率。 5）利用 TopologyBuilder创建 Topology，所有任务执行流程被完整地封装在Topology中。 6）向 Storm集群提交 Topology，任务的执行情况可以通过Nimbus节点的 UI端口实时监控。 3 实验与分析搭建的集群中有 1个 Nimbus节点和 6 个 Supervisor节点，通过高速交换机形成一个192.168.1.0网段的内部网络，在集群无其他任务运行的条件下进行实验。本实验的主要目标是测试Storm的处理性能，主要指标有数据处理延迟、吞吐量、内存使用情况等。定义的 Tuple中包含数据和时间戳两

17、部分，数据部分包含10000个数据对象，时间戳字段在计算数据处理延迟时使用，每经过一个处理模块，就加上当时的时间戳，最后根据这些时间信息计算出处理延迟。 Storm框架支持多种语言接口，选择常用的Java语言分别实现 DBSCAN的普通计算程序和 Storm处理程序，然后运行分析实验结果。 3.1 Storm集群吞吐能力分析龙源期刊网 http:/首先对比分析 Storm集群与 Java 单机环境的吞吐能力，吞吐量反映的是系统在单位时间内处理数据的规模。不断增加任务执行的数据量，记录处理完成所需的时间，为了提高实验结果的准确性，每项数据都是经3 次测试所得的平均值，对比结果见图4。从图 4 可

18、看出：当数据规模较小时，利用Storm集群计算需要更多的时间，这是因为在集群中任务的分发、数据的传输都要经由网络，需要耗费部分系统资源和时间。但随着数据规模的増大，Storm集群的处理能力逐渐显现，处理时间明显缩短。这是因为在 Strom中计算任务被分割成了不同的组件（Spout和 Bolt），由不同的 executor在多个worker中执行。如果按照传统的处理思路，在数据规模稍大时已需要较长时间，可以预见数据规模进一步扩大时，传统求解方法的耗时将难以接受，而Storm集群还可以进一步扩展，满足更大数据量的处理需求。 3.2 数据处理延迟分析单纯对比吞吐量，Storm相对 Hadoop等批处

19、理系统无优势可言。Storm专注于实时数据处理，数据处理的低延迟才是Storm在大数据处理系统中脱颖而出的关键。数据处理延迟在 Storm框架中的定义是从一个 Tuple发出到其被完全处理的时间，Storm提供了 Thrift接口，可以通过编写程序监控任务执行过程中的处理延迟。数据处理延迟和处理模块的并行任务数有关，并行任务数越多，Tuple等待被处理的时间就越短，处理延迟越小，下面通过实验对比说明。将 Spout组件数目设置为 2，DBSCAN计算的 Bolt组件 SingleDbscan数目分别设置为 2 和6，记录两次任务执行时数据处理延迟情况，分别如图56所示。从图 56 可看出：当

20、DBSCAN计算的 Bolt数目设为 2时，数据处理延迟越来越大，这是因为 Spout不断产生新数据，聚类模块无法及时处理掉数据，导致数据积累量越来越多，处理延迟也呈上涨趋势。当DBSCAN计算 Bolt数目为 6时，绝大部分的处理延迟都在毫秒级，表明 Spout产生的数据可以很快得到处理，偶尔有一些处理延迟较大，分析认为是由于任务调度等引起的发送线程在某一刻获取了较多资源，从而发送了较多的数据，但是后端的处理能力较强，可以很快将这些数据处理完毕，数据处理延迟没有明显增长。合理设置各组件的任务数目是优化Storm性能的重要途径，要根据实际的计算任务合理分配，最大限度发挥 Storm的并行处理能

21、力。 3.3 集群内存使用情况 Storm集群中的数据处理都在内存中进行，内存使用是反映集群负载的主要指标。在任务执行期间利用 nmon工具监控集群，各机器的内存使用如图7 所示。龙源期刊网 http:/由图 7 可看出，各节点的内存使用平稳上升，没有出现某台机器内存占用过高的情况。虽然各台机器的物理内存配置不同，在任务执行期间内存使用情况相对合理，负载比较均衡，这是因为 Nimbus节点可以通过 Zookeeper监控各机器的运行状态，根据负载情况分配任务，动态调整各节点的负载。 4 结语针对海量数据的实时响应需求，引入Storm分布式实时处理平台。首先搭建Storm分布式集群，然后设计并编

22、程实现了基于Storm框架的 DBSCAN算法，最后测试分析集群的吞吐量和处理延迟等性能，集群运行状况良好，负载均衡，结果验证了Storm可以胜任大规模数据的实时处理任务。参考文献： 1 JIN R， YANG G， AGRAWAL G. Shared memory parallelization of data miningalgorithms techniques programming interface and performanceJ. IEEE Transactions on Knowledgeand Data Engineering，2005，17（1）：71-89. 2 CUI

23、 J， LI T， LAN H. Design and development of the mass data storage platform based onHadoop J. Journal of Computer Research and Development， 2012， 49（S1）： 12-18.（崔杰，李陶深， 兰红星. 基于 Hadoop的海量数据存储平台设计与开发 J. 计算机研究与发展，2012， 49（增刊 1）： 12-18.） 3 ANDERSON Q. Storm realtime processing cookbookM. Birmingham： Packt

24、 Publishing，2013. 4 EKANAYAKE J， PALLICKARA S. MapReduce for data intensive scientific analysisC/eScience 2008： Proceedings of the Fourth IEEE International Conference on eScience. Piscataway ：IEEE Press， 2008： 277-284. 5 ZHOU Q， CHEN L， MA J， et al. Improved DPhyp algorithm based Impala queryoptimi

25、zationJ. Journal of Computer Research and Development， 2013，50（S2）： 114-120.（周强，陈岭，马骄阳，等.基于改进 DPhyp算法的 Impala查询优化J. 计算机研究与发展，2013，50（S2）： 114-120.） 6 WU G. Spark： the light of big data eraJ. Programmer， 2013（7）：100-104.（吴甘沙.Spark： 大数据的“电光石火”J. 程序员， 2013（7）： 100-104.） 7 LEIBIUSKY J， EISBRUCH G， SIMON

26、ASSI D. Getting started with stormM.Sebastopol： OReilly Media， 2012.龙源期刊网 http:/ 8 YE D， ZHANG R， QIAN X. Marking method for cell fluorescent images based on densityextraction J. Journal of Tsinghua University： Science and Technology， 2013（1）： 129-132.（叶大田，张锐，钱翔.基于密度提取的细胞荧光图像标识算法J. 清华大学学报：自然科学版， 201

27、3 （1）： 129-132.） 9 HOU R， ZHU B， FENG M. Prediction model for lightning now casting based onDBSCANJ. Journal of Computer Applications，2012，32（3）：847-851.（侯荣涛，朱斌，冯民学. 基于 DBSCAN聚类算法的闪电临近预报模型J.计算机应用，2012，32（3）：847-851.） 10 OUYANG J， LIN P. Webpage content extraction based on DBSCAN J. ComputerEngineeri

28、ng，2011，37（3）：64-69.（欧阳佳， 林丕源.基于 DBSCAN算法的网页正文提取J.计算机工程，2011，37（3）：64-69.） 11 LI J， HU X. Efficient mixed clustering algorithm and its application in anomaly detectionJ. Journal of Computer Applications，2010，30（7）：1916-1918.（李建国，胡学钢.高效的混合聚类算法及其在异常检测中的应用J.计算机应用，2010，30（7）：1916-1918.） 12 HAN J W， KAMB

29、ER M， PEI J. Data mining： concepts and techniques M. 3rd ed.San Francisco： Morgan Kaufmann Publishers， 2011. 13 PAN Y， LI G， XU Y. Abnormal data detection method for environment wireless sensornetworks based on DBSCAN J. Computer Applications and Software，2012，29（11）：69-72.（潘渊洋，李光辉，徐勇军. 基于 DBSCAN 的环境传感器网络异常数据检测方法J.计算机应用与软件，2012，29（11）：69-72.）