Spark机器学习——K-Means聚类算法-摩杜云开发者社区

Spark机器学习——K-Means聚类算法_apache

K-Means介绍

中心思想：事先确定常数K，常数K意味着最终的聚类类别数，首先随机选定初始点为质心，并通过计算每一个样本与质心之间的相似度(这里为欧式距离)，将样本点归到最相似的类中。接着，重新计算每个类的质心(即为类中心)，重复这样的过程，直到质心不再改变，最终就确定了每个样本所属的类别以及每个类的质心。由于每次都要计算所有的样本与每一个质心之间的相似度，故在大规模的数据集上，K-Means算法的收敛速度比较慢。

聚类算法：是一种典型的无监督学习算法，主要用于将相似的样本自动归到一个类别中。聚类算法与分类算法最大的区别是：聚类算法是无监督的学习算法，而分类算法属于监督的学习算法，分类是知道结果的。在聚类算法中根据样本之间的相似性，将样本划到不同的类别中，对于不同的相似度计算方法，会得到不同的聚类结果，常用的相似度计算方法有欧式距离法。

K-Means算法的工作流程

聚类分析试图将相似的对象归入同一簇，将不相似的对象归为不同簇，那么，显然需要一种合适的相似度计算方法，我们已知的有很多相似度的计算方法，比如欧氏距离，余弦距离，汉明距离等。事实上，我们应该根据具体的应用来选取合适的相似度计算方法。

当然，任何一种算法都有一定的缺陷，没有一种算法时完美的，有的只是人类不断追求完美，不断创新的意志。K-means算法也有它的缺陷，但是我们可以通过一些后处理来得到更好的聚类结果。

K-means算法虽然比较容易实现，但是其可能收敛到局部最优解，且在大规模数据集上收敛速度相对较慢。

工作流程：

选择聚类的个数k（kmeans算法传递超参数的时候，只需设置最大的K值）
任意产生k个聚类，然后确定聚类中心，或者直接生成k个中心
对每个点确定其聚类中心点
再计算其聚类新中心
重复以上步骤直到满足收敛要求。（通常就是确定的中心点不再改变。）

首先，随机确定k个初始点的质心；然后将数据集中的每一个点分配到一个簇中，即为每一个点找到距其最近的质心，并将其分配给该质心所对应的簇；该步完成后，每一个簇的质心更新为该簇所有点的平均值。伪代码如下：

创建k个点作为起始质心，可以随机选择(位于数据边界内)
	当任意一个点的簇分配结果发生改变时
		对数据集中每一个点
			对每个质心
				计算质心与数据点之间的距离
			将数据点分配到距其最近的簇
		对每一个簇，计算簇中所有点的均值并将均值作为质心

需要说明的是，在算法中，相似度的计算方法默认的是欧氏距离计算，当然也可以使用其他相似度计算函数，比如余弦距离；算法中，k个类的初始化方式为随机初始化，并且初始化的质心必须在整个数据集的边界之内，这可以通过找到数据集每一维的最大值和最小值；然后最小值+取值范围*(0到1的随机数)，来确保随机点在数据边界之内。

在实际的K-means算法中，采用计算质心-分配-重新计算质心的方式反复迭代，算法停止的条件是当然数据集所有的点分配的距其最近的簇不在发生变化时，就停止分配，更新所有簇的质心后，返回k个类的质心(一般是向量的形式)组成的质心列表，以及存储各个数据点的分类结果和误差距离的平方的二维矩阵。

上面返回的结果中，之所以存储每个数据点距离其质心误差距离平方，是便于后续的算法预处理。因为K-means算法采取的是随机初始化k个簇的质心的方式，因此聚类效果又可能陷入局部最优解的情况，局部最优解虽然效果不错，但不如全局最优解的聚类效果更好。

算法优化之二分K-Means算法

为得到更好的结果，可以尝试采用优化后的二分K-Means算法。二分K-means算法首先将所有点作为一个簇，然后将簇一分为二。之后选择其中一个簇继续进行划分，选择哪一个簇取决于对其进行划分是否能够最大程度的降低SSE的值。上述划分过程不断重复，直至划分的簇的数目达到用户指定的值为止。二分K-means算法的伪代码如下：

将所有点看成一个簇
当簇数目小于k时
对于每一个簇
	计算总误差
	在给定的簇上面进行k-均值聚类（k=2）
	计算将该簇一分为二之后的总误差
选择使得总误差最小的簇进行划分

K-Means之Spark实例

官方案例链接

http://spark.apache.org/docs/2.2.2/mllib-clustering.html

attack.txt

team,score,goal,shoot,straight,violated,offside
拜仁,18,24,150,65,54,13
热刺,10,18,80,41,60,8
巴黎,16,17,75,35,90,16
皇马,11,14,121,47,73,6
曼城,14,16,92,39,53,15
亚特兰大,7,8,87,32,76,11
尤文图斯,16,12,79,33,83,9
马竞,10,8,96,35,74,14
利物浦,13,13,104,36,39,8
那不勒斯,12,11,87,22,77,10
巴塞罗那,14,9,82,33,79,8
多特蒙德,10,8,73,29,56,11
莱比锡红牛,11,10,105,35,51,16
里昂,8,9,87,28,71,13
瓦伦西亚,11,9,70,26,115,9
切尔西,11,11,108,37,62,8

Maven依赖

<properties>
    <scala.version>2.11.8</scala.version>
    <spark.version>2.2.2</spark.version>
    <hadoop.version>2.7.6</hadoop.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-mllib_2.11</artifactId>
        <version>${spark.version}</version>
    </dependency>
</dependencies>

KMeans.scala

package ml

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.clustering.KMeans

/**
  * @Author Daniel
  * @Description 基于kMeans聚类算法的足球队实力分析
  *
  **/
object KMeans {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("KMeans")
      .getOrCreate()
    val df = spark.read
      // 设置源数据格式
      .format("com.databricks.spark.csv")
      // 一次只读一行数据
      .option("inferSchema", "true")
      .option("header", true)
      .load("attack.txt")
    //    attackDf.show(30, false)
    // 注册一个udf来将数据转换为向量
    spark.udf
      .register("toVector", (
                              a: Double, b: Double, c: Double,
                              d: Double, e: Double, f: Double
                            ) => Vectors.dense(a, b, c, d, e, f)
      )
    // 创建特征向量
    val features = df.selectExpr("toVector(score,goal,shoot,straight,violated,offside) as feature", "team")
    // 生成K-Means模型,将数据分为两类
    val kMeans = new KMeans().setK(3)
      .setFeaturesCol("feature")
      .setPredictionCol("prediction")
    val model = kMeans.fit(features)
    // 计算聚合中心点
    val predicted = model.transform(features)
    //    predicted.show(30, false)
    predicted.createTempView("predicted")
    // 分组显示结果
    val sql =
      """
        |SELECT *
        |FROM predicted
        | ORDER BY prediction DESC
      """.stripMargin
    val group = spark.sql(sql)
    group.show(30, false)
  }
}

输出结果

+--------------------------------+-----+----------+
|feature                         |team |prediction|
+--------------------------------+-----+----------+
|[14.0,16.0,92.0,39.0,53.0,15.0] |曼城   |2         |
|[11.0,11.0,108.0,37.0,62.0,8.0] |切尔西  |2         |
|[13.0,13.0,104.0,36.0,39.0,8.0] |利物浦  |2         |
|[11.0,14.0,121.0,47.0,73.0,6.0] |皇马   |2         |
|[11.0,10.0,105.0,35.0,51.0,16.0]|莱比锡红牛|2         |
|[18.0,24.0,150.0,65.0,54.0,13.0]|拜仁   |1         |
|[10.0,18.0,80.0,41.0,60.0,8.0]  |热刺   |0         |
|[16.0,17.0,75.0,35.0,90.0,16.0] |巴黎   |0         |
|[12.0,11.0,87.0,22.0,77.0,10.0] |那不勒斯 |0         |
|[7.0,8.0,87.0,32.0,76.0,11.0]   |亚特兰大 |0         |
|[16.0,12.0,79.0,33.0,83.0,9.0]  |尤文图斯 |0         |
|[14.0,9.0,82.0,33.0,79.0,8.0]   |巴塞罗那 |0         |
|[10.0,8.0,73.0,29.0,56.0,11.0]  |多特蒙德 |0         |
|[11.0,9.0,70.0,26.0,115.0,9.0]  |瓦伦西亚 |0         |
|[10.0,8.0,96.0,35.0,74.0,14.0]  |马竞   |0         |
|[8.0,9.0,87.0,28.0,71.0,13.0]   |里昂   |0         |
+--------------------------------+-----+----------+