Spark 2.1.0 入门：KMeans聚类算法(Python版)

KMeans 是一个迭代求解的聚类算法，其属于划分（Partitioning）型的聚类方法，即首先创建K个划分，然后迭代地将样本从一个划分转移到另一个划分来改善最终聚类的质量。

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

1.根据给定的k值，选取k个样本点作为初始划分中心；
2.计算所有样本点到每一个划分中心的距离，并将所有样本点划分到距离最近的划分中心；
3.计算每个划分中样本点的平均值，将其作为新的中心；

循环进行2~3步直至达到最大迭代次数，或划分中心的变化小于某一预定义阈值
显然，初始划分中心的选取在很大程度上决定了最终聚类的质量，和MLlib包一样，ML包内置的KMeans类也提供了名为 KMeans|| 的初始划分中心选择方法，它是著名的 KMeans++ 方法的并行化版本，其思想是令初始聚类中心尽可能的互相远离，具体实现细节可以参见斯坦福大学的B Bahmani在PVLDB上的论文Scalable K-Means++，这里不再赘述。

与MLlib版本的Kmeans教程相同，本文亦使用UCI数据集中的鸢尾花数据Iris进行实验，它可以在iris获取，Iris数据的样本容量为150，有四个实数值的特征，分别代表花朵四个部位的尺寸，以及该样本对应鸢尾花的亚种类型（共有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
...

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

from pyspark.sql import Row
from pyspark.ml.clustering import KMeans,KMeansModel
from pyspark.ml.linalg import Vectors

下文中，我们默认名为spark的SparkSession已经创建。

rawData = sc.textFile("file:///usr/local/spark/iris.txt")
def f(x):
    rel = {}
    rel['features'] = Vectors.dense(float(x[0]),float(x[1]),float(x[2]),float(x[3]))
    return rel

df = sc.textFile("file:///usr/local/spark/iris.txt").map(lambda line: line.split(',')).map(lambda p: Row(**f(p))).toDF()

在得到数据后，我们即可通过ML包的固有流程：创建Estimator并调用其fit()方法来生成相应的Transformer对象，很显然，在这里KMeans类是Estimator，而用于保存训练后模型的KMeansModel类则属于Transformer：

kmeansmodel = KMeans().setK(3).setFeaturesCol('features').setPredictionCol('prediction').fit(df)

与MLlib版本类似，ML包下的KMeans方法也有Seed（随机数种子）、Tol（收敛阈值）、K（簇个数）、MaxIter（最大迭代次数）、initMode（初始化方式）、initStep（KMeans||方法的步数）等参数可供设置，和其他的ML框架算法一样，用户可以通过相应的setXXX()方法来进行设置，或以ParamMap的形式传入参数，这里为了简介期间，使用setXXX()方法设置了参数K，其余参数均采用默认值。

与MLlib中的实现不同，KMeansModel作为一个Transformer，不再提供predict()样式的方法，而是提供了一致性的transform()方法，用于将存储在DataFrame中的给定数据集进行整体处理，生成带有预测簇标签的数据集：

results = kmeansmodel.transform(df).collect()
for item in results:
...     print(str(item[0])+' is predcted as cluster'+ str(item[1]))
... 
[5.1,3.5,1.4,0.2] is predcted as cluster1
[4.9,3.0,1.4,0.2] is predcted as cluster1
[4.7,3.2,1.3,0.2] is predcted as cluster1
[4.6,3.1,1.5,0.2] is predcted as cluster1
[5.0,3.6,1.4,0.2] is predcted as cluster1
[5.4,3.9,1.7,0.4] is predcted as cluster1
[4.6,3.4,1.4,0.3] is predcted as cluster1
[5.0,3.4,1.5,0.2] is predcted as cluster1
[4.4,2.9,1.4,0.2] is predcted as cluster1
[4.9,3.1,1.5,0.1] is predcted as cluster1
[5.4,3.7,1.5,0.2] is predcted as cluster1
[4.8,3.4,1.6,0.2] is predcted as cluster1
[4.8,3.0,1.4,0.1] is predcted as cluster1
[4.3,3.0,1.1,0.1] is predcted as cluster1
[5.8,4.0,1.2,0.2] is predcted as cluster1

也可以通过KMeansModel类自带的clusterCenters属性获取到模型的所有聚类中心情况：

results2 = kmeansmodel.clusterCenters()
for item in results2:
...     print(item)
... 
[ 5.9016129   2.7483871   4.39354839  1.43387097]
[ 5.006  3.418  1.464  0.244]
[ 6.85        3.07368421  5.74210526  2.07105263]

与MLlib下的实现相同，KMeansModel类也提供了计算集合内误差平方和（Within Set Sum of Squared Error, WSSSE) 的方法来度量聚类的有效性，在真实K值未知的情况下，该值的变化可以作为选取合适K值的一个重要参考：

kmeansmodel.computeCost(data)
78.94084142614622

厦大数据库实验室博客