实战Mahout聚类算法Canopy+K-means

2019年3月21日 209次阅读来源: 聚类算法

转载：实战Mahout聚类算法Canopy+K-means

原文来自：http://my.oschina.net/BreathL/blog/58104

Mahout是Apache的顶级开源项目，它由Lucene衍生而来，且基于Hadoop的，对处理大规模数据的机器学习的经典算法提供了高效的实现。其中，对经典的聚类算法即提供了单机实现，同时也提供了基于hadoop分布式的实现，都是非常好的学习资料。

聚类分析

聚类（Clustering）可以简单的理解为将数据对象分为多个簇（Cluster），每个簇里的所有数据对象具有一定的相似性，这样一个簇可以看多一个整体对待，以此可以提高计算质量或减少计算量。而数据对象间相似性的衡量有不少经典算法可以用，但它们所需的数据结构基本一致，那就是向量；常见的有欧几里得距离算法、余弦距离算法、皮尔逊相关系数算法等，Mahout对此都提供了实现，并且你可以在实现自己的聚类时，通过接口切换不同的距离算法。

数据模型

在Mahout的聚类分析的计算过程中，数据对象会转化成向量（Vector）参与运算，在Mahout中的接口是org.apache.mahout.math.Vector 它里面每个域用一个浮点数（double）表示，你可以通过继承Mahout里的基类如：AbstractVector来实现自己的向量模型，也可以直接使用一些它提供的已有实现如下：

1. DenseVector，它的实现就是一个浮点数数组，对向量里所有域都进行存储，适合用于存储密集向量。

2. RandomAccessSparseVector 基于浮点数的 HashMap 实现的，key 是整形 (int) 类型，value 是浮点数(double) 类型，它只存储向量中不为空的值，并提供随机访问。

3. SequentialAccessVector 实现为整形 (int) 类型和浮点数 (double) 类型的并行数组，它也只存储向量中不为空的值，但只提供顺序访问。

聚类算法K-means与Canopy

首先介绍先K-means算法：所有做聚类分析的数据对象，会被描述成n为空间中的一个点，用向量（Vector）表示；算法开始会随机选择K个点，作为一个簇的中心，然后其余的点会根据它与每个簇心的距离，被分配到最近簇中去；接着以迭代的方式，先重新计算每个簇的中心（通过其包含的所有向量的平均值），计算完成后对所有点属于哪个簇进行重新划分；一直如此迭代直到过程收敛；可证明迭代次数是有限的。

虽然K-means简单且高效，但它存在一定问题，首先K值（即簇的数量）是人为确定的，在对数据不了解的情况下，很难给出合理的K值；其次初始簇心的选择是随机的，若选择到了较孤立的点，会对聚类的效果产生非常大的影响。因此通常会用Canopy算法配合，进行初始化，确定簇数以及初始簇心。

Canopy算法首先会要求输入两个阀值 T1和T2，T1>T2；算法有一个集群这里叫Canopy的集合（Set），当然一开始它是空的；然后会将读取到的第一个点作为集合中的一个Canopy，接着读取下一个点，若该点与集合中的每个Canopy计算距离，若这个距离小于T1，则这个点会分配给这个Canopy（一个点可以分配给多个Canopy），而当这个距离小于T2时这个点不能作为一个新的Canopy而放到集合中。也就是说当一个点只要与集合中任意一个Canopy的距离小于T2了，即表示它里那个Canopy太近不能作为新的Canopy。若都没有则生成一个新的Canopy放入集合中。以此循环，直到没有点了。

所以这里用到的聚类分析算法的思路是：首先通过Canopy算法进行聚类，以确定簇数以及初始簇心的，接着通过K-means算法进行迭代运算，收敛出最后的聚类结果。接下来我们看看实现。

实战

首先需要Java的环境不用多说，我这用的JDK1.6；同时还需要搭建Hadoop分布式系统，网上有很多帖子，这里也不细讲，我的版本是2.0.2。

接着是安装Mahout，你可以通过svn获取 http://svn.apache.org/repos/asf/mahout/trunk 但需要有maven的支持，你也可以直接下载源码 http://www.apache.org/dyn/closer.cgi/mahout/ 。推荐使用svn的方式；我这里通过svn获取源码后，通过Maven进行编译，生成如下项目：

简单说明下： mahout-core：核心程序模块；mahout-math：在核心程序中使用的一些数据通用计算模块；mahout-utils：在核心程序中使用的一些通用的工具性模块；最后 mahout-examples 是Mahout提供的一些实现的例子，可作为使用Mahout进行编程的非常好的参考，我们的例子也从这里来。

在 mahout-examples 中的 org.apache.mahout.clustering.syntheticcontrol.kmeans.Job类，对上述算法提供了较完整的实现，它是一个Hadoop的job，我们从源代码入手，看如何将实际的数据跑起来。下面是该类的核心逻辑代码：

`01`	`public` `static` `void` `run(Configuration conf, Path input, Path output,`

`02`	`DistanceMeasure measure,` `double` `t1,` `double` `t2,` `double` `convergenceDelta,`

`03`	`int` `maxIterations)`

`04`	`throws` `Exception{`

`05`	`Path directoryContainingConvertedInput =` `new` `Path(output,`

`06`	`DIRECTORY_CONTAINING_CONVERTED_INPUT);`

`07`	`log.info("Preparing Input");`

`08`	`InputDriver.runJob(input, directoryContainingConvertedInput,`

`09`	`"org.apache.mahout.math.RandomAccessSparseVector");`

`10`	`log.info("Running Canopy to get initial clusters");`

`11`	`CanopyDriver.run(conf, directoryContainingConvertedInput, output, measure,`

`12`	`t1, t2,` `false,` `false);`

`13`	`log.info("Running KMeans");`

`14`	`KMeansDriver.run(conf, directoryContainingConvertedInput,` `new` `Path(output,`

`15`	`Cluster.INITIAL_CLUSTERS_DIR), output, measure, convergenceDelta,`

`16`	`maxIterations,` `true,` `false);`

`17`	`// run ClusterDumper`

`18`	`ClusterDumper clusterDumper =` `new` `ClusterDumper(finalClusterPath(conf,`

`19`	`output, maxIterations),` `new` `Path(output,` `"clusteredPoints"));`

`20`	`clusterDumper.printClusters(null);`

21 }

这个例子中调用了3个Map/Reduce 任务以及一个转换，它们如下：

1. 第8行： InputDriver.runJob ( ) ，它用于将原始数据文件转换成 Mahout进行计算所需格式的文件 SequenceFile，它是Hadoop API提供的一种二进制文件支持。这种二进制文件直接将<key, value>对序列化到文件中。

2. 第11行：CanopyDriver.run( ) ，即用Canopy算法确定初始簇的个数和簇的中心。

3. 第14行：KMeansDriver.run( ) ，这显然是K-means算法进行聚类。

4. 第18~20行，ClusterDumper类将聚类的结果装换并写出来，若你了解了源代码，你也可以自己实现这个类的功能，因为聚类后的数据存储格式，往往跟自身业务有关。

这里细讲下第一个Map/Reduce： InputDriver.runJob ( )因为我们需要了解，初始数据的格式，其他的任务CanopyDriver.run( )和KMeansDriver.run( )任务就不细讲了，主要就是Canopy和K-means算法，原理已经介绍了，实现也不难，需要你了解hadoop编程。

InputDriver.runJob( )实现也非常简单，它只有Map，其代码如下：

`01`	`@Override`

`02`	`protected` `void` `map(LongWritable key, Text values, Context context)` `throws` `IOException, InterruptedException {`

03

`04`	`String[] numbers = SPACE.split(values.toString());`

`05`	`// sometimes there are multiple separator spaces`

`06`	`Collection<Double> doubles = Lists.newArrayList();`

`07`	`for` `(String value : numbers) {`

`08`	`if` `(!value.isEmpty()) {`

`09`	`doubles.add(Double.valueOf(value));`

10 }

11 }

`12`	`// ignore empty lines in data file`

`13`	`if` `(!doubles.isEmpty()) {`

14 try {

`15`	`Vector result = (Vector) constructor.newInstance(doubles.size());`

`16`	`int` `index =` `0;`

`17`	`for` `(Double d : doubles) {`

`18`	`result.set(index++, d);`

19 }

`20`	`VectorWritable vectorWritable =` `new` `VectorWritable(result);`

`21`	`context.write(new` `Text(String.valueOf(index)), vectorWritable);`

22

`23`	`}` `catch` `(InstantiationException e) {`

`24`	`throw` `new` `IllegalStateException(e);`

`25`	`}` `catch` `(IllegalAccessException e) {`

`26`	`throw` `new` `IllegalStateException(e);`

`27`	`}` `catch` `(InvocationTargetException e) {`

`28`	`throw` `new` `IllegalStateException(e);`

29 }

30 }

31 }

由代码可以看出，它将你初始数据文件的每一行用空格切开成个 String[] numbers ，然后再将 numbers中的每个String转换成Double类型，并以此生成一个向量 Vector ，然后通过 SequenceFileOutputFormat的方式输出成SequenceFile，以作下一步计算的输入。由此我们可以了解到我们的初始数据的格式需要以一行为一个单位，用空格分隔，每一列为一个Double数即可（当然你也可以反过来修改例子中的实现）。

如此准备好初始数据后，我们将mahout-examples 编译并打成jar ，若你有用maven的话，它会自动帮你加载其所依赖的jar，若没用，请确保不少jar，然后将该jar包放到搭建了hadoop环境的服务器上（Linux操作系统），我没改名字，jar包叫mahout-examples-0.7-SNAPSHOT-job.jar。

好了，一切准备就绪，我们可以在服务器上运行如下命令，来运行聚类任务：

`1`	`hadoop jar ../mahout-examples-0.7-SNAPSHOT-job.jar org.apache.mahout.clustering.syntheticcontrol.kmeans.Job`

`2`	`-i <数据输入路径> \`

`3`	`-o <数据输出路径> \`

`4`	`-k <K值> \`

`5`	`-cl <是否先使用Canopy 算法初始化` `true` `or` `false>`

`6`	`-t1 <Canopy中的T1值>`

`7`	`-t2 <Canopy中的T2值>`

`8`	`-xm <执行方式: sequential（单机运行） or mapreduce（分布式）>`

在运行完成之后，数据由于是SequenceFile存储还无法查看，需要使用 Mahout 的 ClusterDump 程序转储聚类质心（和相关的点）。最终结果将存储在 kmeans 目录下名称以 clusters- 开头、以 -final 结尾的子目录中。具体的值将取决于运行任务使用了多少次迭代，例如 clusters-2-final 是第三次迭代的输出结果。可执行下面命令进行转换。

`1`	`hadoop jar /soft/mahout/mahout/examples/target/mahout-examples-0.7-SNAPSHOT-job.jar org.apache.mahout.utils.clustering.ClusterDumper -i ..(上一步的输出路径)/clusters-2-final/ --pointsDir ..(上一步输出路径)/clusteredPoints -o <要保持的文件名>`

最后你就可以在你指定的文件中查看聚类分析结果了。

参考资料：

https://cwiki.apache.org/confluence/display/MAHOUT/K-Means+Clustering

https://cwiki.apache.org/confluence/display/MAHOUT/Canopy+Clustering

http://www.ibm.com/developerworks/cn/java/j-mahout-scaling/

http://www.ibm.com/developerworks/cn/web/1103_zhaoct_recommstudy3/

《Mahout in action》

https://cwiki.apache.org/MAHOUT/cluster-dumper.html