基于Spark的并行KM eans聚类模型研究∗

侯敬儒　吴　晟　李英娜

（昆明理工大学信息工程与自动化学院　昆明　650500）

1　引言

聚类分析是在没有给定具体分类或者clusters（群组）的情况下，通过探索性计算数据之间指定属性上的相识性的一种无监督统计分析学习的过程，然后将数据划分为相交或不相交簇［1］。聚类作为数据挖掘领域中一种分析工具，已经在包括经济学、生物学、信息检索等许多领域广泛应用［2］。大数据时代背景下，互联网信息爆炸式增长，数据量、数据类型等产生速度越来越迅猛［3］，越来越大的数据规模等待着聚类分析计算，而作为一种有效且常用的KMeans均值聚类算法，它单机执行的时间复杂度比较高［4］，所以其处理能力有一定的局限性。

Hadoop是目前广泛使用的并行计算平台［5］，但Hadoop的MapReduce并不适合迭代式计算。在Hadoop的MapReduce计算模型中，一个任务只有map映射和reduce化简两个计算阶段［6］，复杂的计算任务需要设计为多个Job共同完成，各个Job之间的依赖关系是由MapReduce设计开发者自己进行管理的；而且map映射阶段的中间结果需要写到本地磁盘当中去，这对需要多次迭代才能完成的计算是不适合的。而迭代式计算在数据处理科学中是经常见到的，特别是在Data Mining、Information Retrieval、Machine Learning等领域［7］，大部分算法都是通过多次迭代来计算完成的。

新一代分布式大数据并行处理框架Spark，弥补了Hadoop在迭代式计算方面的不足。它是一个基于内存计算的开源集群计算系统，是由加州大学伯克利分校AMP实验室使用Scala语言开发的，目前已经是Apache的顶级开源项目，是Apache社区最火热的项目之一，并且已经成为发展最快的大数据处理平台之一［8］。文章论述了使用KMeans算法在Spark分布式计算平台的并行实现原理并且给出了其实现过程。

2　相关研究

2.1　大数据相关研究

在大数据分布式计算主流技术中，Hadoop依靠廉价PC［9］研究设计了分布式并行存储、计算框架，实现了分布式集群的高可靠、高吞吐、易扩展性，很快发展成为大数据分布式科学计算中的重要技术，它主要以分布式存储框架HDFS和分布式计算框架MapReduce为核心组件，汇集和派生出许多周围的数据采集、集群管理、数据分析应用等工具。但是，分布式计算框架MapReduce在程序runtime时需要在集群的各个Worker节点进行data-copy，从而消耗大量时间在网络和磁盘I/O读写上，而不是在任务计算上。所以，Hadoop的MapReduce架构在Stream Computing（流式计算）和Ad Hoc（即席查询）等应用场景中不太适用［10］。

2.2　Spark计算框架研究

Apache Spark是一个分布式计算框架，旨在简化运行于计算机集群上的并行程序的编写。该框架对资源调度，任务的提交、执行和跟踪，节点间的通信以及数据并行处理的内在底层操作都进行了抽象。它提供了一个更高级别的API用于处理分布式数据。从这方面说，它与Apache Hadoop的MapReduce分布式计算框架类似。但在底层架构上，Spark与它有所不同。

Spark起源于加利福尼亚大学伯克利分校AMP实验室的一个研究项目［11］。AMP实验室当时关注分布式机器学习算法的应用情况。因此，Spark从一开始便为应对迭代式计算应用的高性能需求而设计。在这类应用中，相同的数据会被多次访问。该设计主要靠利用数据集内存缓存以及启动任务时的低延迟和低系统开销来实现高性能。再加上其容错性、灵活的分布式数据结构和强大的函数式编程接口，Spark在各类基于机器学习和迭代式分析的大规模数据处理任务上有广泛的应用，这也表明了其实用性。

3　基于Spark的并行KM eans聚类模型

3.1　KM eans算法

传统串行的KMeans均值算法的思想一般是先初始化随机给定K个clusters（簇）中心，将待分类的样本数据点依照最近邻原则分配到各个cluster中。之后以一定法则重新计算各个cluster的质心，以确定新的cluster。不停的循环计算，直到cluster的质心移动距离小于某一个实际给定的具体值。

传统KMeans均值聚类算法“三步走”如下简介：

1）寻找待聚类的样本数据点的聚类中心。

2）计算每个样本数据点到之前寻找到的聚类中心的距离，将每个样本数据点聚类到离该样本数据点最近的聚类中去。

3）计算每个聚类中所有样本数据点坐标的平均值，然后将这个计算的平均值作为新的聚类中心。

循环执行第2）、3）步，一直计算到新的聚类中心移动距离小于某个给定的具体值或者聚类次数达到了设定值为止［12］。

3.2　KM eans算法的并行化

根据以上对KMeans算法的描述，同时结合Spark平台的并行编程模型，设计了基于Spark的KMeans并行聚类模型方案，如图1所示。在程序运行之前，将模型训练数据集上传到HDFS中去。Spark集群的任务调度系统DAGScheduler的Task-Scheduler则会为分割后每个数据子集在Executor里创建新job，并利用分布式集群的资源调度器给相应的job分配计算资源。然后模型在Spark集群上进行并行KMeans训练，训练完成后，输出全局最优模型到HDFS分布式文件存储系统。

图1　基于Spark的并行KMeans聚类模型

3.2.1生成初始聚类中心点

对于初始化聚类中心点，我们可以在输入的数据集中随机地选取k个点作为中心点，但是随机选择初始中心点可能会造成聚类的结果和数据的实际分布相差很大。文章使用KMeans++算法选择聚类初始中心点。

KMeans++算法选择初始聚类中心点的基本思想为：初始的聚类中心之间的坐标距离要尽可能远，过程如下。

1）从读取到的样本数据点中随机选择一个样本点作为第一个聚类中心；

2）对于读取到的样本数据集中的每一个样本数据点x，计算它与最近聚类中心的坐标距离D(x)；

3）选择新的样本数据点作为新的聚类中心，规则是选择D(x)较大的样本数据点作为聚类中心；

迭代执行第2）、3）步骤，一直等到K个聚类中心通过距离计算被选择出来；然后用这K个被选出来的初始聚类中心开始并行运行KMeans算法。

从上面的算法描述可以看到，算法的关键是第3）步，如何将D(x)反映到点被选择的概率上。文章采用算法如下。

1）随机从点集D中选择一个点作为初始的中心点；

2）计算每一个点到最近中心点的距离Si，对所有Si求和得到sum；

3）然后再取一个随机值，用权重的方式计算下一个“种子点”。取随机值random(0＜random＜sum)，对点集 D循环，做random-=Si运算，直到random＜0，那么点i就是下一个中心点。

重复2）和3），直到 K个聚类中心被选出来。利用这K个初始的聚类中心来运行KMeans算法。

3.2.2迭代计算样本的中心点

迭代计算中心点的并行实现：首先计算每个样本属于哪个中心点，之后采用聚合函数统计属于每个中心点的样本值之和以及样本数量，然后求得最新中心点，并且判断中心点是否发生改变。其中对于计算样本属于哪个中心点，采用一种快速查找、计算距离的方法，其方法如下。

首先定义lowerBoundOfSqDist距离公式，假设中心点 center是(a1，b1)，需要计算的点 point是(a2，b2)，那么 lowerBoundOfSqDist是：

可轻易证明lowerBoundOfSqDist将会小于或等于EuclideanDist，因此在进行距离比较的时候，先计算很容易计算的lowerBoundOfSqDist（只需要计算center、point的L2范数）。如果lowerBoundOfSqDist都不小于之前计算得到的最小距离bestDistance，那真正的欧氏距离也不可能小于bestDistance了。因此在这种情况下就不需要去计算欧氏距离了，省去了很多计算工作。

如果lowerBoundOfSqDist小于bestDistance，则进行距离的计算，调用 fastSquaredDistance，该方法是一种快速计算距离的方法。fastSquaredDistance方法会先计算一个精度，有关精度的计算：

precisionBound1=2.0*EPSILON*

sumSquaredNorm/(normDiff*normDiff+EPSILON)

如果精度满足条件，则欧氏距离为EuclideanDist=sumSquaredNorm-2.0*v1.dot(v2)，其中sumSquaredNorm为+++，2.0*v1.dot(v2)为2(a1a2+b1b2)。这里可以直接利用之前计算的L2范数。如果精度不满足要求，则进行原始的距离计算，公式为(a1-a2)2+(b1-b2)2。

4　实验过程与结果分析

4.1　实验环境与数据集

文章实验基于QJM（Quorum Journal Manager）下的HA（High Available）大数据平台，其中，以Hadoop的HDFS为基础存储框架，主要以Spark为计算框架，使用Zookeeper统筹HA下的大数据平台，管理整个集群配置。平台包括2个Master节点和4个Worker节点，节点之间局域网连接；平台的资源管理和任务调度采用Yarn模式。软件配置：平台搭建使用spark-1.6.3-bin-hadoop2.6和hadoop-2.6.5，Scala选用 2.10.5版本，Java选 用JDK1.8.0_101（Linux版），操作系统为Centos6.5。

文章使用MovieLens数据集：Ml-100k（含10万条评分集）、Ml-1m（含100万条评分集）、ml-10m100k（含1000万条评分集）。

4.2　实验结果与性能分析

4.2.1模型训练过程分析

文章基于KMeans并行聚类模型编写的Spark Application主要包含两个过程：1）加载数据集；2）模型训练。整个程序被拆分为33个job，54个stage，432个task。图2是执行过程中所完成的jobs：包含每一个由Spark的Action操作触发的jo-bid、触发job的action操作详细信息、job提交时间、jobs间隔时间、该job被分为几个stage、该job的所有stages中包含的task数量。

图2　Spark Application jobs执行过程

基于Spark的KMeans模型训练过程中的几个具有代表性的stage的DAG如图3所示。从左至右依次为数据采样stage、各个分区数据聚合stage、collect操作stage、collectAsMap操作stage。

图3　KMeans模型的Stage-DAG拓扑结构

4.2.2模型训练结果与数据partition的关系

为了评估模型在不同分区下训练时间的变化情况，选用了MovieLens数据集的Ml-10m100k（1000万）进行实验，每个分区在数据集上的训练时间取3次训练的平均时间，最终结果如图4所示。

图4　分区数对运行时间的影响

从图4可以看到，随着分区数目的增加，模型的训练时间先是急剧减少随后缓慢增加。这是由于分区数目变大导致算法的并行度提高，虽然降低了分区的计算时间，但同时也增加了分区数据传输的网络通信负担。开始时partition分区数量的增加会明显降低每个partition分区的jobs计算时间，进而减少整个程序的运行时间；但是partition分区数达到一定的量以后，单个partition分区的jobs计算时间趋于稳定，而网络通信负担则会随着partition分区数量的增大而迅速增加，所以训练时间则会呈现出小幅度增加的态势。

4.2.3算法运行时间对比

为了评估算法在不同节点下训练时间的变化情况，选用了MovieLens数据集的Ml-100k（10万）、ml-1m（100万）、Ml-10m100k（1000万）进行实验，最终结果如图5所示。

图5　算法运行时间对比

图5记录对比了模型在不同规模数据集、不同节点数量上的运行时间。当数据集较小时单机（图5横轴为0的是单机版运行时间，横轴为1的是含有一个节点的集群运行时间，以此类推）的KMeans要比在基于Spark集群（一个worker节点）环境下运行KMeans算法训练时间少，这是因为集群计算需要初始化相关任务，包括job的新建、资源的分配和调度，而且不同的任务之间也需要通信，这都是需要消费时间成本的。随着数据规模的增大，节点数量的增加，基于Spark的并行KMeans模型在集群上运行的优势也就越明显，产生这种结果是因为在大规模数据下训练KMeans聚类模型的任务初始化和通信时间占整个作业运行时间的比重较小，同时该并行KMeans算法会将样本训练数据在加载到集群的时候就分割为固定大小的Data Block数据块在不同Worker、不同处理器上并行计算，从而节省大量训练时间。

5　结语

文章基于Spark分布式计算框架，在基于QJM的HA大数据平台下，设计并实现了并行KMeans聚类模型，通过模型训练实验表明，该分布式聚类模型适合运行在大数据分布式平台下；并且通过repartition算子分片加载数据集优化了并行方案。

在接下来的学习中，将对基于Spark的在线KMeans聚类模型展开研究，实现流数据的实时KMeans聚类。

［1］刘泽燊，潘志松.基于Spark的并行SVM算法研究［J］.计算机科学，2016，43（5）：238-242.LIU Zeshen，PAN Zhisong.Research on parallel algorithm for the SVM based on Spark［J］.Computer science，2016，43（5）：238-242.

［2］Alsheikh MA，Niyato D，Lin S等.基于深度学习和Spark的移动大数据分析［J］.IEEE 网络，2016，30（3）：22-29.Alsheikh MA，Niyato D，Lin S，etal.Mobile big data analytics using deep learning and apache spark［J］.IEEE Network，2016，30（3）：22-29.

［3］孟建良，刘德超.一种基于Spark和聚类分析的辨识电力系统不良数据新方法［J］.电力系统保护与控制，2016，44（3）：85-91.MENG Jianliang，LIU Dechao.A new method for identifying bad data of power system based on Spark and clustering analysis［J］.Power System Protection and Control，2016，44（3）：85-91.

［4］Bosagh Zadeh R，Meng X，Ulanov A等.基于Spark的矩阵计算与优化［J］.2016：31-38.Bosagh Zadeh R，Meng X，Ulanov A，et al.Matrix Computations and Optimization in Apache Spark［J］.2016：31-38.

［5］Marcu O C，Costan A，Antoniu G，等.Spark vs Flink：理解大数据分析框架性能［C］//IEEE国际集群计算会议.IEEE计算机协会，2016：433-442.Marcu O C，Costan A，Antoniu G，et al.Spark Versus Flink：Understanding Performance in Big Data Analytics Frameworks［C］//IEEE International Conference on CLUSTER Computing.IEEEComputer Society，2016：433-442.

［6］孙科.基于Spark的机器学习应用框架研究与实现［D］.上海：上海交通大学，2015.SUN Ke.Research and implementation ofmachine learning application framework on spark［D］.Shanghai：Shanghai Jiao Tong University，2015.

［7］Shkapsky A，Yang M，Interlandi M，等.基于 Spark的Datalog查 询 大 数 据 分 析［C］//国 际 会 议 ，2016：1135-1149.Shkapsky A，Yang M，InterlandiM，etal.Big Data Analytics with Datalog Queries on Spark［C］//International Conference，2016：1135-1149.

［8］Liu J，Liang Y，AnsariN.基于Spark的大数据处理之大规模矩阵求逆［J］.IEEEAccess，2016，4：1-1.Liu J，Liang Y，Ansari N.Spark-Based Large-Scale Matrix Inversion for Big Data Processing［J］.IEEE Access，2016，4：1-1.

［9］宁永恒.基于Spark的若干数据挖掘技术研究［D］.杭州：中国计量学院，2015.NING Yongheng.A number of datamining technology research based on the Spark［D］.Hangzhou：Journal of China Jiliang University，2015.

［10］黎文阳.大数据处理模型Apache Spark研究［J］.现代计算机：普及版，2015（3）：55-60.LIWenyang.Research on Apache Spark for Big Data Processing［J］.Modern Computer，2015（3）：55-60.

［11］萨初日拉，周国亮，时磊，等.Spark环境下并行立方体计算方法［J］.计算机应用，2016，36（2）：348-352.SACHURILA，ZHOU Guoliang，SHI Lei，等.Parallel cube computing in Spark［J］.Journalof Computer Applications，2016，36（2）：348-352.

［12］陈梦杰，陈勇旭，贾益斌，等.基于Hadoop的大数据查询系统简述［J］.计算机与数字工程，2013，41（12）：1939-1942.CHENMengjie，CHEN Yongxu，JIA Yibin，etal.A Brief Introduction Hadoop—based Big Data Query System［J］.Computer&Digital Engineering，2013，41（12）：1939-1942.