基于流处理改进的SK-means策略

姜晓艳，张 伟,2，王佳慧，马利民

(1.北京信息科技大学 计算机学院，北京 100101；2.北京信息科技大学 北京材料基因工程高精尖创新中心，北京 100101；3.国家信息中心，北京 100038)

0 引言

互联网、物联网的快速发展[1]，数据采集和传输技术的进步，促使大量且连续的动态流数据产生。要获取流数据中的信息，需要对其进行有效的聚类和可持续分析。K-means作为一种经典的聚类算法，在各个领域得到了广泛应用。但是在大规模数据场景下，K-means聚类算法存在总体速度较慢、聚类准确度较低的问题。

针对上述问题，学者们尝试基于大数据计算框架对K-means聚类算法的初始聚类中心选择及并行加速两方面进行研究。文献[2]提出了一种利用随机最大最小距离选择初始聚类中心的算法，并使用批处理框架MapReduce进行实现，但算法仍存在迭代次数多导致耗时长的局限性。文献[3]提出了一种基于密度峰值的初始聚类中心选择算法，并在批处理框架Spark中实现，但由于批处理框架本身的局限性，聚类耗时优化效果不显著。

以上研究的技术方案都是以批处理框架[4]为基础，相较于流处理，批处理计算时延较长，实时性较差，随着时间的流逝，获取到的信息的价值会逐渐降低。

目前，以流处理框架[5]为基础优化K-means算法的研究相对较少。文献[6]提出了一种以聚类中心之间的距离为标准，优化初选聚类中心的方案，其使用流处理框架Apache Flink[7]实现操作算子并行化加速算法的执行，但仍要求在数据全部读入后才能开始计算，导致数据读入的过程耗时占比较大，且没有针对流处理的特点对聚类过程进行改进，聚类时间仍然较长。

本文提出了一种基于流处理思想的SK-means(streamK-means)策略，并通过Apache Flink框架予以实现，旨在提高传统K-means算法的聚类速度。

1 基于流处理的SK-means策略

1.1 K-means算法模型

根据数据集之间的相似指标,K-means聚类算法将数据分为若干簇，首先随机地指定K个数据点作为初始聚类中心，计算各数据点之间的相似度，并根据相似度进行分类，然后更新聚类中心，计算各数据点的相似度重新分类，当误差平方总和收敛或小于指定阈值或迭代次数达到上限停止[8]。算法流程如图1所示。

图1 K-means算法流程

假设现有数据点个数为N，预计聚类数量为K，原始数据集用向量X表示，向量C1、C2、…、Ck分别表示K个聚类中心，其中k∈{1,2,…,K}。每个数据样本包含M个特征。数据集X表示如下：

算法具体步骤如下：

1)选取初始聚类中心。从全部数据中随机或采用某种方案选择出K个数据点作为初始的聚类中心，用C1、C2、…、Ck表示。为了减少聚类整体计算时间，采用流处理的思想，在数据读入的同时选取初始聚类中心。初始聚类中心集合向量C为

2)计算样本相似性。计算数据点与已有类之间的相似度，选取相似度最大的类为该数据点的类。其中式(1)为选用欧氏距离计算相似度，式(2)为选取最大相似度即最小欧氏距离。

n∈{1,2,…,N},m∈{1,2,…,M}

(1)

Xi={min(fEUD(xn,Ck))}

i∈{1,2,…,N}

(2)

3)通过不断地迭代步骤1)、2)直到聚类中心与前一次计算结果之间的变化量小于给定的阈值为止。

1.2 基于流处理思想改进K-means算法模型

流处理可以不间断接收数据流，不再需要等待数据全部读完之后触发事件，而是不断更新结果，非常适合用于对实时数据进行处理。为了降低算法整体耗时，本文提出了一种基于流处理思想选取初始聚类中心的方法。该算法与基于点距离判断改进的K-means算法[9]有两点明显不同：第一，算法无需等待全部数据到达之后才开始计算距离相似度，从有数据流入行为开始，算法就可以同步执行计算；第二，算法并非用来改进K-means聚类迭代过程，而是用来计算K-means算法所需要的输入参数(初始中心和类的数量)，且在新数据点加入某类时，并不重新计算当前类的中心，从而最大程度减少初选聚类中心的繁琐步骤，缩短初选聚类中心的时间。

由1.1中算法分析可知，算法全过程的整体耗时可以分为4个部分[10]：获取数据源时间，确定初始聚类中心时间，聚类时间和结果数据输出时间。从文献[11]中可以看出，获取数据源时间在总体聚类过程所用时间中占比较大，约为46%，等待全部数据读入结束之后，才计算初始聚类中心，导致整体聚类过程的耗时长。

传统K-means算法中获取数据源和确定初始聚类中心两阶段是先后关系，不能同时进行，必须等到上一阶段全部完成，下一阶段才能开始。为此，本文从流处理思想出发，利用Flink框架流处理的特性，实现在数据读入的过程中计算初始中心的功能，如图2所示。一方面，基于时间复用的原则，最大程度地减少获取数据源和确定初始聚类中心两阶段耗时；另一方面，输出高质量初始中心，以减少K-means迭代聚类阶段的迭代次数，进而降低整体耗时。

图2 SK-means各阶段耗时

初始聚类中心计算的具体优化步骤如下:

1)从读入的数据流中选取第一个数据点D1，作为第一个聚类中心，加入到初始聚类中心的集合中，记为O1，并设定距离阈值T。如图3所示，以O1为圆心，T为半径的圆形区域为O1的有效范围，两个有效范围重合地方的样本为公有样本。

图3 初始聚类中心样本模型

2)如图4所示，计算数据流中下一个数据点Di到已有聚类中心集合O中所有聚类中心的相似度距离dk，其中黑色点为中心，白色点为新到达的数据。若存在某点使得dk

图4 聚类中心集合不变的情况

否则，认为第k数据点不属于任何一个已存在的聚类中心的有效范围，此时需要更新聚类中心集合，将点Di加入到聚类中心集合O中，作为一个新的聚类中心。依次对每个数据流执行以上操作，直到数据流读入结束为止，如图5所示。

图5 聚类中心集合更新的情况

3)第二步产生的聚类中心集合即为K-means算法所需要的初始聚类中心集合，同时可以通过集合中聚类中心的数量确定出K值。

优化之后的确定初始聚类中心的方法与K-means结合，本文简称为SK-means策略，其优势如下：首先，将不在已有聚类中心的有效范围内的数据即时添加到聚类中心集合中，能够快速呈现出新发现的聚类中心，用于之后读入的数据流的计算；其次，不需要预先对聚类的数量做假设，而是随着数据流的读入对其不断更新；最后，在读入数据的同时进行初始聚类中心的确定，减少聚类过程整体耗时。

2 SK-means策略实现

2.1 Apache Flink平台

Apache Flink是一个既可以实现流处理又可以实现批处理功能的通用型的开源、高性能、分布式的大数据框架[12]。不同于传统的流处理平台，例如Apache Storm[13]，Flink支持有状态的流处理和Exactly-Once恰好一次的容错机制[14]，并在流处理基础上实现了批处理功能。因此，Flink得到了学术界的广泛关注。

框架整体由主节点和计算节点两种类型节点组成。如图6、图7 所示，主节点上运行的是JobManager服务，它是Flink计算集群的核心服务，负责给TaskManager分配任务，并实现错误恢复以及管理检查点等。任务被提交后，信息会汇集到这里；计算节点运行的是TaskManager服务，它会一直监听JobManager的信息，并执行被JobManager分配的任务。两者之间通过心跳机制相互联系，互换任务信息，以便及时对任务进行调整调度。JVM进程的多个线程中可以同时运行多个TaskManager，从而实现灵活、并发执行任务[15]。

图6 Flink基础架构

图7 Flink计算框架

Flink的执行过程可以抽象成有向无环图，如图8所示，主要分为3部分，首尾分别为Source读入和Sink输出，其中最重要的逻辑部分为转换(Transformation)操作，也是方案实现时比重最大、最困难的部分。图8为将读入和操作两部分的并行度设定为2、输出部分并行度设定为1时的示例流程图。具体并行度可根据实际情况进行更改。

图8 Flink并行化数据处理示例流程

2.2 基于Flink平台的SK-means策略实现

SK-means算法使用流处理框架Apache Flink平台来实现并行化加速。与已有的少数Flink平台实现方案不同，本文使用了Flink的流处理API、广播等特有机制来实现边读边计算的优化方案，减少不必要的等待耗时。

如前文所述，Flink 应用程序可以分为数据源读入、转换操作和数据输出3个阶段。在读入阶段，从Hadoop分布式文件系统(HDFS)中读入原始数据，HDFS能够将一个大的文件分成若干块分布式存储，以实现Flink多数据源读取。在转换操作阶段，若干算子使用广播状态进行数据的传递。在K-means算法中，需要经常动态地修改处理聚类中心集合的情况，非常适合使用广播状态进行处理。聚类中心集合被广播到另一个Operator的所有并发实例中，被保存为状态。当输入数据流读入时会发给同一个Operator的各个实例，并与广播的聚类中心集合一起计算。在输出阶段，将计算结果输出到分布式文件系统HDFS中，以便后续使用。

Flink平台实现SK-means算法首先要解析命令行参数，构建执行环境，然后在读入数据的同时计算出初始聚类中心集合，以此聚类中心集合作为K-means算法的输入进行计算，最后将结果进行输出。主流程伪代码如下：

算法1SK-means

params = ParameterTool.fromArgs(args);

env =getExecutionEnvironment();

centroids = getDataAndPoint(params,env);

clusteredPoints =K-means(centroids);

clusteredPoints.write();

env.execute();

主要优化getDataAndPoint数据读入时确定初始聚类中心过程，对读入的每一个点执行转换操作。首先从广播中获取聚类中心集合pointDataSet，然后遍历聚类中心，判断数据流中数据点到聚类中心的相似度距离是否小于阈值T。若存在一个聚类中心满足条件，则直接返回，若不存在，则将该数据点加入到初始聚类中心集合中，更新聚类集合，该点成为新的聚类中心。改进后算法的伪代码如下：

算法2getDataAndPoint

input:points,params

centroids = points.map({

pointData= getBroadcastVariable("pointData");

for(Point centroid：pointData)

if(centroid.getDistance(value)

returnnull;

pointData.add(value);

returnvalue;

});

output：centroids；

3 实验与分析

3.1 实验环境

本实验环境由1个主节点和2个计算节点共3台服务器组成，数据源点负责数据的输入，汇点负责存储相关性能指标信息以及聚类结果。服务器硬件参数详情如表1所示，节点软件配置详情如表2所示。Hadoop[16]开源框架可以用于分布式地存储大规模数据。由于其高容错性和可扩展性，在早期的大数据处理中被大量使用。为了使集群达到最优的性能，根据现有软硬件环境，对Flink分布式集群的相关配置参数进行调整，其中重要的参数配置项机器参数值如表3所示。

表1 服务器硬件参数详情表

表2 节点软件配置参数详情表

表3 集群配置参数详情表

3.2 评价标准和数据集

本实验使用调整的兰德指数对聚类结果进行评价。假设模型的超分布为随机模型，且不需要对聚类的结构进行任何假设，通常用于比较各种聚类算法效果(例如K-means)。兰德指数的值越大，表明实验效果越好。

实验使用来自UCI机器学习数据库的Iris数据集，这是研究者们常用的公认数据集。Iris数据集总大小为2.21 kB，有150条样本，每类50个数据，共3类，每个样本包含4个特征字段。

为了对比SK-means与K-means算法的效果，实验需要较大体量的数据集。借鉴文献[17]中的数据模拟方法，在基础数据集点周围生成多倍的额外点来扩展数据集，选取Iris数据集随机生成大小为215.44 MB共1.5×107条记录进行测试。

此外，在算法执行过程中，使用“聚类中心变量变化小于某阈值”这一条件作为迭代终止条件，且阈值设为极小的10-6，确保结果的精确性。

3.3 实验结果

3.3.1 SK-means与K-means算法效果对比

分别使用Apache Flink实现K-means算法和优化后的SK-means算法对数据进行分类，并对比迭代次数和兰德指数，取10次实验平均值,结果如表4所示。由表可知，相较于K-means聚类算法，SK-means算法聚类的迭代次数较少，兰德指数更高，聚类效果更好。

表4 迭代次数与兰德指数对比

3.3.2 SK-means算法耗时测试

由于算法主要解决的问题是算法聚类耗时长，因此，对优化之后算法的整体耗时情况进行测试，以验证优化效果。由Apache Flink官方监控界面可以看到任务的执行情况及计算耗时。在此基础上，多次实验取耗时平均值如表5所示。

表5 不同并行度下的耗时 s

可以看出，SK-means算法在不同并行度下的整体耗时都有所减少，整体性能有所提升。分析原因是传统K-means算法受输入的初始聚类中心的随机性影响较大，当初始聚类中心质量较差时，会导致整体耗时较长。实验过程中存在输入特定初始聚类中心集合的情况下，传统K-means耗时低于SK-means算法的情况。但是，由于SK-means直接根据当前读入的数据流确定初始聚类中心，所以，当输入不同初始聚类中心对传统K-means算法进行多次实验时，传统K-means算法耗时略高于SK-means算法。

从表5中的测试数据可以得出，在并行度分别为4、8、12时，SK-means算法的整体计算耗时较传统K-means算法耗时分别减少了18.89%、18.52%、3.39%，平均减少13.6%，优化效果明显。

4 结束语

针对大数据流式计算框架下，数据规模过大时聚类算法K-means会出现的聚类时间长、需要预先指定初始聚类中心的问题，提出了一种基于大数据计算框架Flink并且结合流处理思想进行优化的SK-means算法。使用流处理的思想，在数据读入的过程中确定初始聚类中心，并基于改进后的SK-means算法，在Apache Flink平台实现并行化加速SK-means策略。最后从迭代次数、兰德指数、整体耗时3个方面设计实验，结果表明，优化后的SK-means算法能够有效减少聚类时的迭代次数，减少聚类计算整体耗时，提高算法的执行效率，提升聚类效果。