《Hadoop与大数据挖掘》——2.5　K-Means算法原理及Hadoop MapReduce实现

本节书摘来自华章计算机《Hadoop与大数据挖掘》一书中的第2章，第2.5节，作者 张良均 樊哲 位文超 刘名军 许国杰 周龙 焦正升，更多章节内容可以访问云栖社区“华章计算机”公众号查看。

2.5　K-Means算法原理及Hadoop MapReduce实现

2.5.1　K-Means算法原理

K-Means算法是硬聚类算法，是典型的基于原型的目标函数聚类方法的代表。它是将数据点到原型的某种距离作为优化的目标函数，利用函数求极值的方法得到迭代运算的调整规则（如图2-45所示）。K-Means算法以欧氏距离作为相似度测度，求对应某一初始聚类中心向量V最优分类，使得评价指标最小。算法采用误差平方和准则函数作为聚类准则函数。

具体的算法步骤如下：

2.5.2　动手实践：K-Means算法实现

编写单机版的K-Means算法有利于理解Hadoop实现的K-Means算法，所以这里给出单机版（Java）的编写步骤，供读者参考。

实验步骤如下：

1）打开Eclipse，新建Java工程kmeans1.0；
2）参考前面的流程完善K-means代码；
3）使用测试数据hadoop/data/kmeans.data进行测试，查看结果；
4）思考把该算法转换为Hadoop MapReduce实现的思路。

2.5.3　Hadoop K-Means算法实现思路

针对K-Means算法，本节给出两种实现思路。思路1相对比较直观，但是效率较低；思路2在实现上需要自定义键值类型，但是效率较高。下面是对两种思路的介绍。

思路1

如图2-47所示，算法描述如下：

1）根据原始文件生成随机聚类中心向量（需指定聚类中心向量个数k），指定循环次数；
2）在map阶段，setup函数读取并初始化聚类中心向量；在map函数中读取每个记录，计算当前记录到各个聚类中心向量的距离，根据到聚类中心向量最小的聚类中心id判断该记录属于哪个类别，输出所属聚类中心id和当前记录；
3）在reduce阶段，reduce函数接收相同聚类中心id的数据；把这些数据的每列进行求和，并记录每列的个数；计算新的聚类中心向量（每列的和除以每列的个数），然后输出聚类中心id和新的聚类中心向量；
4）判断前后两次聚类中心向量之间的误差是否小于某阈值；如果小于，则跳转到步骤5），否则跳转到步骤2）；
5）针对最后一次生成的聚类中心向量对原始数据进行分类，得到每个记录的类别。

其MR数据流如图2-48所示。

思路2

如如图2-49所示，算法描述如下：

1）根据原始文件生成随机聚类中心向量（需指定聚类中心向量个数k），指定循环次数。
2）在map阶段，setup函数读取并初始化聚类中心向量，同时初始化聚类中心向量和；在map函数中读取每个记录，计算当前记录到各个聚类中心向量的距离，根据到聚类中心向量最小的聚类中心id判断该记录属于哪个类别，然后把所属的类别加入到聚类中心向量和中（需要记录个数及和，即需要自定义类型）；在cleanup函数中输出所属聚类中心id和其对应的聚类中心向量和。

3）在reduce阶段，reduce函数接收相同聚类中心id的数据；把这些数据的每列进行求和，并记录每列的个数；计算新的聚类中心向量（每列的和除以每列的个数），然后输出聚类中心id和新的聚类中心向量。
4）判断前后两次聚类中心向量之间的误差是否小于某阈值；如果小于，则跳转到步骤5），否则跳转到步骤2）。
5）针对最后一次生成的聚类中心向量对原始数据进行分类，得到每个记录的类别。

其MR数据流如图2-50所示。

2.5.4　Hadoop K-Means编程实现

在下面的实现过程中，会进行简单实现思路介绍，针对一些实现会有动手实践给读者练习。一般情况下我们建议读者自己全部实现，对于实现起来有难度的读者，我们提供了参考程序，但是需要注意，参考程序不是完整的，里面设置了TODO提示，这些地方是需要读者去完善的。

不管是思路1还是思路2，Hadoop实现K-Means算法都包含4个步骤：①初始化聚类中心向量；②进行聚类并更新聚类中心向量；③判断是否达到循环条件，如果是则循环；④判断是否需要对原始数据进行分类，如果是则进行分类操作。下面就针对这4个步骤分别进行分析。

（1）初始化聚类中心向量：蓄水池抽样

初始化聚类中心其实和单机算法类似，可以有多种方法，比如随机取出k个聚类中心向量、直接取出前k个聚类中心向量等。在Hadoop的编程框架MapReduce限制下，如果是随机取k个聚类中心向量，那么实现起来就是这样的：遍历一次所有数据，统计数据个数n，再次遍历，按照k/n概率抽取k个数据。这样不是不可以，但是效率太低，并且如果真要实现起来，还是要考虑多个问题的，比如如果有多个Mapper怎么处理？

这里提出一种效率高，并且还能达到随机取数的算法—蓄水池抽样。

什么是蓄水池抽样呢？简单描述：先选中第1～k个元素，作为被选中的元素。然后依次对第k＋1至第n个元素做如下操作：每个元素都有k/x的概率被选中，然后等概率地（1/k）替换掉被选中的元素（其中x是元素的序号）。其算法伪代码描述如代码清单2-33所示。

蓄水池抽样同样可以使用Driver、Mapper、Reducer来进行分析。Driver部分可以参考MapReduce程序的固定模式，但是需要注意，需要传入聚类中心向量的个数，即k值。其代码参考代码清单2-34。

Mapper就是蓄水池抽样算法的具体实现了，这里需要注意，map函数针对每条记录进行筛选，并不输出，所以这里在cleanup进行输出。这样就需要在setup里面初始化一个变量来存储当前已经被选为聚类中心向量的值。其各个函数描述如下。

在设计Reducer的时候需要考虑的一个问题是，如果有多个Mapper怎么办？多个Mapper就会发送k×N个聚类中心向量到Reducer中（其中N为Mapper的个数），所以在Reducer端需要对k×N个记录再次筛选，选出其中的k个聚类中心向量。这里当然也有多种方法，其实这里的选择和最开始我们在Mapper中针对所有数据随机选取k条记录的选择一样，这里所有数据只是“变”小了而已。因为是在Reducer中处理（一个Reducer可以理解为单机），所以其实也可以理解为单机的随机选择k条记录的算法。这里随机选择k条记录的算法也可以，不过我们这里还是选择使用蓄水池抽样。

这里只能使用一个Reducer，为什么？请读者思考。

动手实践：蓄水池抽样Hadoop实现

首先理解上面蓄水池抽样算法的Hadoop实现的描述及分析，接着新建工程，并参考上节完善工程代码功能。

实验步骤：

1）打开Eclipse，新建工程2.5_002_sample；
2）添加相关环境（如JDK路径、Hadoop路径等）；
3）参考上节蓄水池抽样Hadoop实现原理实现编写源代码；
4）把工程编译，并导出jar包，然后上传jar包到master节点上，使用yarn jar的方式运行，查看输出及相关日志。

思考：

1）还有其他方式实现蓄水池抽样吗？
2）如何查看蓄水池抽样抽取出来的结果？
（2）更新聚类中心向量
更新聚类中心向量其实就是整个K-Means算法的核心所在，K-Means算法的每次循环其实就是一个不断更新聚类中心向量的过程。那么具体怎么更新呢？我们在单机算法中已经知道怎么更新了，怎么把其转换为Hadoop的MapReduce代码呢？其实，可以把每个Mapper理解为一个单机算法，因为其处理的数据其实是所有数据的一部分（一个文件块）。下面来看具体涉及的Driver、Mapper和Reducer。

针对Driver类，除了一些固定写法外，还需传入聚类初始中心向量路径、聚类中心个数、列分隔符（考虑是否需要？），其示例代码如代码清单2-36所示。

Reducer设置多个会有什么问题？可以设置多个吗？设置多个有什么好处？

Mapper的工作主要包括两个：其一，读取首次HDFS上的聚类中心；其二，根据聚类中心对每个键值对记录进行距离计算，输出距离最小的聚类中心ID以及该条键值对记录。下面针对具体实现做分析。

1）setup()：读取传入的初始聚类中心向量路径，根据路径读取对应的数据，利用分隔符来对初始聚类中心向量进行初始化（初始化为数组和列表）。
2）map()：在map阶段根据初始化的聚类中心向量对当前记录进行分类，输出其对应的聚类中心id、当前记录，如代码清单2-37所示。

要做的工作就是针对每个组的所有数据计算其平均值（该平均值就是新的聚类中心向量）。其函数描述如下。

1）reduce()：每个reduce函数针对同一个聚类中心id的数据进行处理；具体处理过程为，把每条记录对应列的值加起来，同时记录当前的记录数；接着，使用每列和除以记录数，即可得到每列平均值，也就是当前聚类中心id新的聚类中心，如代码清单2-38所示。

3）cleanup()：输出每个类别新的聚类中心。

动手实践：Hadoop实现更新聚类中心向量

实验步骤如下：

1）打开Eclipse，打开上一小节完成的工程；
2）根据上节Hadoop实现更新聚类中心实现思路，编写对应源代码；
3）把工程编译并导出Jar包，然后上传Jar包到master节点上，使用yarn jar的方式运行，查看输出及相关日志。

思考：如何测试代码？

同时，需要覆写readFields、write函数，在这里针对数组类型还需要做些额外的处理。其处理过程为存储数组的长度，在实例化类的时候传入数组的长度，否则会报NullPointer的异常，如代码清单2-40所示。

写入或者读取时，注意顺序，顺序重要吗？如果乱序会有什么影响？请读者思考。

下面针对Mapper进行分析。

SumNumWritable构造函数如代码清单2-42所示。

Reducer只需要整合各个Mapper的输出记录，针对每个记录分别求“列和”、个数和，然后再求平均即可得到新的聚类中心向量和。各个函数描述如下。

（2）动手实践：Hadoop实现K-Means算法思路2

请读者参考思路1的动手实践，编写K-Means算法思路2的Hadoop实现。