一 KMeans簡介

KMeans 是一個迭代求解的聚類算法，其屬于 劃分（Partitioning） 型的聚類方法，即首先創(chuàng)建K個劃分，然后迭代地將樣本從一個劃分轉移到另一個劃分來改善最終聚類的質量。

ML包下的KMeans方法位于org.apache.spark.ml.clustering包下，其過程大致如下：

1.根據(jù)給定的k值，選取k個樣本點作為初始劃分中心；
2.計算所有樣本點到每一個劃分中心的距離，并將所有樣本點劃分到距離最近的劃分中心；
3.計算每個劃分中樣本點的平均值，將其作為新的中心；
循環(huán)進行2~3步直至達到最大迭代次數(shù)，或劃分中心的變化小于某一預定義閾值

顯然，初始劃分中心的選取在很大程度上決定了最終聚類的質量，和MLlib包一樣，ML包內置的KMeans類也提供了名為 KMeans|| 的初始劃分中心選擇方法，它是著名的 KMeans++ 方法的并行化版本，其思想是令初始聚類中心盡可能的互相遠離，具體實現(xiàn)細節(jié)可以參見斯坦福大學的B Bahmani在PVLDB上的論文Scalable K-Means++，這里不再贅述。

二 實戰(zhàn)

與MLlib版本的Kmeans教程相同，本文亦使用UCI數(shù)據(jù)集中的鳶尾花數(shù)據(jù)Iris進行實驗，它可以在iris獲取，Iris數(shù)據(jù)的樣本容量為150，有四個實數(shù)值的特征，分別代表花朵四個部位的尺寸，以及該樣本對應鳶尾花的亞種類型（共有3種亞種類型），如下所示：

5.1,3.5,1.4,0.2,setosa
...
5.4,3.0,4.5,1.5,versicolor
...
7.1,3.0,5.9,2.1,virginica
...

在使用前，引入需要的包：

import org.apache.spark.ml.clustering.{KMeans,KMeansModel}
import org.apache.spark.ml.linalg.{Vector,Vectors}

開啟RDD的隱式轉換：

import spark.implicits._

下文中，我們默認名為spark的SparkSession已經(jīng)創(chuàng)建。

為了便于生成相應的DataFrame，這里定義一個名為model_instance的case class作為DataFrame每一行（一個數(shù)據(jù)樣本）的數(shù)據(jù)類型。
注：因為是非監(jiān)督學習，所以不需要數(shù)據(jù)中的label,只需要使用特征向量數(shù)據(jù)就可以。

scala> case class model_instance(features: org.apache.spark.ml.linalg.Vector)
defined class model_instance

在定義數(shù)據(jù)類型完成后，即可將數(shù)據(jù)讀入RDD[model_instance]的結構中，并通過RDD的隱式轉換.toDF()方法完成RDD到DataFrame的轉換：

val rawData = sc.textFile("file:///root/data/iris.txt")
rawData: org.apache.spark.rdd.RDD[String] = file:///root/data/iris.txt MapPartitionsRDD[220] at textFile at <console>:56scala> val df = rawData.map(line =>{ model_instance( Vectors.dense(line.split(",").filter(p => p.matches("\\d*(\\.?)\\d*")).map(_.toDouble)) )}).toDF()
df: org.apache.spark.sql.DataFrame = [features: vector]

與MLlib版的教程類似，我們使用了filter算子，過濾掉類標簽，正則表達式\\d*(\\.?)\\d*可以用于匹配實數(shù)類型的數(shù)字，\\d*使用了*限定符，表示匹配0次或多次的數(shù)字字符，\\.?使用了?限定符，表示匹配0次或1次的小數(shù)點。

在得到數(shù)據(jù)后，我們即可通過ML包的固有流程：創(chuàng)建Estimator并調用其fit()方法來生成相應的Transformer對象，很顯然，在這里KMeans類是Estimator，而用于保存訓練后模型的KMeansModel類則屬于Transformer：

scala> val kmeansmodel = new KMeans().setK(3).setFeaturesCol("features").setPredictionCol("prediction").fit(df)
kmeansmodel: org.apache.spark.ml.clustering.KMeansModel = kmeans_bdadcf53b52e

與MLlib版本類似，ML包下的KMeans方法也有Seed（隨機數(shù)種子）、Tol（收斂閾值）、K（簇個數(shù)）、MaxIter（最大迭代次數(shù)）、initMode（初始化方式）、initStep（KMeans||方法的步數(shù)）等參數(shù)可供設置，和其他的ML框架算法一樣，用戶可以通過相應的setXXX()方法來進行設置，或以ParamMap的形式傳入?yún)?shù)，這里為了簡介期間，使用setXXX()方法設置了參數(shù)K，其余參數(shù)均采用默認值

與MLlib中的實現(xiàn)不同，KMeansModel作為一個Transformer，不再提供predict()樣式的方法，而是提供了一致性的transform()方法，用于將存儲在DataFrame中的給定數(shù)據(jù)集進行整體處理，生成帶有預測簇標簽的數(shù)據(jù)集：

scala> val results = kmeansmodel.transform(df)
results: org.apache.spark.sql.DataFrame = [features: vector, prediction: int]

為了方便觀察，我們可以使用collect()方法，該方法將DataFrame中所有的數(shù)據(jù)組織成一個Array對象進行返回：

scala> results.collect().foreach(row => {println( row(0) + " is predicted as cluster " + row(1))})
[5.1,3.5,1.4,0.2] is predicted as cluster 2
...
[6.3,3.3,6.0,2.5] is predicted as cluster 1
...
[5.8,2.7,5.1,1.9] is predicted as cluster 0
...scala> results.show()
+-----------------+----------+
|         features|prediction|
+-----------------+----------+
|[5.1,3.5,1.4,0.2]|         0|
|[4.9,3.0,1.4,0.2]|         0|
|[4.7,3.2,1.3,0.2]|         0|
|[4.6,3.1,1.5,0.2]|         0|
|[5.0,3.6,1.4,0.2]|         0|
|[5.4,3.9,1.7,0.4]|         0|
|[4.6,3.4,1.4,0.3]|         0|
|[5.0,3.4,1.5,0.2]|         0|
|[4.4,2.9,1.4,0.2]|         0|
|[4.9,3.1,1.5,0.1]|         0|
|[5.4,3.7,1.5,0.2]|         0|
|[4.8,3.4,1.6,0.2]|         0|
|[4.8,3.0,1.4,0.1]|         0|
|[4.3,3.0,1.1,0.1]|         0|
|[5.8,4.0,1.2,0.2]|         0|
|[5.7,4.4,1.5,0.4]|         0|
|[5.4,3.9,1.3,0.4]|         0|
|[5.1,3.5,1.4,0.3]|         0|
|[5.7,3.8,1.7,0.3]|         0|
|[5.1,3.8,1.5,0.3]|         0|
+-----------------+----------+
only showing top 20 rows

也可以通過KMeansModel類自帶的clusterCenters屬性獲取到模型的所有聚類中心情況：

scala> kmeansmodel.clusterCenters.foreach(center => {println("Clustering Center:"+center)})
Clustering Center:[5.005999999999999,3.4180000000000006,1.4640000000000002,0.2439999999999999]
Clustering Center:[5.901612903225806,2.7483870967741932,4.393548387096774,1.433870967741935]
Clustering Center:[6.85,3.0736842105263147,5.742105263157893,2.071052631578947]

與MLlib下的實現(xiàn)相同，KMeansModel類也提供了計算 集合內誤差平方和（Within Set Sum of Squared Error, WSSSE) 的方法來度量聚類的有效性，在真實K值未知的情況下，該值的變化可以作為選取合適K值的一個重要參考：

scala> kmeansmodel.computeCost(df)
res30: Double = 78.94084142614622