使用 Spark MLlib 做 K-means 聚类分析（2）

look_w

MLlib 的 K-means 实现Spark MLlib 中 K-means 算法的实现类 (KMeans.scala) 具有以下参数，具体如下。
图 1. MLlib K-means 算法实现类预览通过下面默认构造函数，我们可以看到这些可调参数具有以下初始值。
图 2. MLlib K-means 算法参数初始值参数的含义解释如下：

• k 表示期望的聚类的个数。
• maxInterations 表示方法单次运行最大的迭代次数。
• runs 表示算法被运行的次数。K-means 算法不保证能返回全局最优的聚类结果，所以在目标数据集上多次跑 K-means 算法，有助于返回最佳聚类结果。
• initializationMode 表示初始聚类中心点的选择方式, 目前支持随机选择或者 K-means||方式。默认是 K-means||。
• initializationSteps表示 K-means||方法中的部数。
• epsilon 表示 K-means 算法迭代收敛的阀值。
• seed 表示集群初始化时的随机种子。

通常应用时，我们都会先调用 KMeans.train 方法对数据集进行聚类训练，这个方法会返回 KMeansModel 类实例，然后我们也可以使用 KMeansModel.predict                                方法对新的数据点进行所属聚类的预测，这是非常实用的功能。
KMeans.train 方法有很多重载方法，这里我们选择参数最全的一个展示。
图 3. KMeans.train 方法预览KMeansModel.predict 方法接受不同的参数，可以是向量，或者 RDD，返回是入参所属的聚类的索引号。
图 4. KMeansModel.predict 方法预览聚类测试数据集简介在本文中，我们所用到目标数据集是来自 UCI Machine Learning Repository 的 Wholesale customer Data                                Set。UCI 是一个关于机器学习测试数据的下载中心站点，里面包含了适用于做聚类，分群，回归等各种机器学习问题的数据集。
Wholesale customer Data Set                                是引用某批发经销商的客户在各种类别产品上的年消费数。为了方便处理，本文把原始的 CSV 格式转化成了两个文本文件，分别是训练用数据和测试用数据。
图 5. 客户消费数据格式预览读者可以从标题清楚的看到每一列代表的含义，当然读者也可以到 UCI 网站上去找到关于该数据集的更多信息。虽然 UCI 的数据可以自由获取并使用，但是我们还是在此声明，该数据集的版权属 UCI 以及其原始提供组织或公司所有。
案例分析和编码实现本例中，我们将根据目标客户的消费数据，将每一列视为一个特征指标，对数据集进行聚类分析。代码实现步骤如下
清单 1. 聚类分析实现类源码

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33

import org.apache.spark.{SparkContext, SparkConf} import org.apache.spark.mllib.clustering.{KMeans, KMeansModel} import org.apache.spark.mllib.linalg.Vectors object KMeansClustering {<br> def main (args: Array[String]) {<br> if (args.length < 5) {<br>     println("Usage:KMeansClustering trainingDataFilePath testDataFilePath numClusters     numIterations runTimes")<br> sys.exit(1)<br> }<br><br> val conf = new     SparkConf().setAppName("Spark MLlib Exercise:K-Means Clustering")<br> val sc = new SparkContext(conf)<br>    /**<br> *Channel Region Fresh Milk Grocery Frozen Detergents_Paper Delicassen<br> * 2 3      12669 9656 7561 214 2674 1338<br> * 2 3 7057 9810 9568 1762 3293 1776<br> * 2 3 6353 8808      7684 2405 3516 7844<br> */<br>     val rawTrainingData = sc.textFile(args(0))<br> val parsedTrainingData =     rawTrainingData.filter(!isColumnNameLine(_)).map(line => {<br>     Vectors.dense(line.split("\t").map(_.trim).filter(!"".equals(_)).map(_.toDouble))<br> }).cache()<br>     // Cluster the data into two classes using KMeans<br>     val numClusters = args(2).toInt<br> val numIterations = args(3).toInt<br> val runTimes =     args(4).toInt<br> var clusterIndex:Int = 0<br> val clusters:KMeansModel =     KMeans.train(parsedTrainingData, numClusters, numIterations,runTimes)<br>     println("Cluster Number:" + clusters.clusterCenters.length)<br>     println("Cluster Centers Information Overview:")<br> clusters.clusterCenters.foreach(     x => {<br>     println("Center Point of Cluster " + clusterIndex + ":")<br>     println(x)<br> clusterIndex += 1<br> })<br>     //begin to check which cluster each test data belongs to based on the clustering result<br>     val rawTestData = sc.textFile(args(1))<br> val parsedTestData = rawTestData.map(line =>     {<br>     Vectors.dense(line.split("\t").map(_.trim).filter(!"".equals(_)).map(_.toDouble))<br>     })<br> parsedTestData.collect().foreach(testDataLine => {<br> val predictedClusterIndex:     Int = clusters.predict(testDataLine)<br>     println("The data " + testDataLine.toString + " belongs to cluster " +     predictedClusterIndex)<br> })<br>     println("Spark MLlib K-means clustering test finished.")<br> }<br><br> private def     isColumnNameLine(line:String):Boolean = {<br> if (line != null &&     line.contains("Channel")) true<br> else false<br> }

该示例程序接受五个入参，分别是

• 训练数据集文件路径
• 测试数据集文件路径
• 聚类的个数
• K-means 算法的迭代次数
• K-means 算法 run 的次数