KMeans是一种典型的聚类算法，本文通过代码来演示用spark运行KMeans算法的一个小例子。

1 算法简介

KMeans算法的基本思想是初始随机给定K个簇中心，按照最邻近原则把无标签样本点分到各个簇。然后按平均法重新计算各个簇的质心，从而确定新的簇心。一直迭代，直到簇心的移动距离小于某个给定的值或迭代次数达到阈值。

2 运行步骤

2.1 数据说明

数据格式为：特征1 特征2 特征3

0.0 0.0 0.0
0.1 0.1 0.1
0.2 0.2 0.2
9.0 9.0 9.0
9.1 9.1 9.1
9.2 9.2 9.2

2.2 代码及说明

import org.apache.log4j.{ Level, Logger }
import org.apache.spark.mllib.clustering.KMeans
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.SparkSession

object KMeansTest {
  def main(args: Array[String]): Unit = {
    // 生成SparkSession对象
    val spark = new SparkSession.Builder().appName("KMeansTest").getOrCreate()
    //生成SparkContext对象
    val sc = spark.sparkContext
    //设置日志输出级别
    Logger.getRootLogger.setLevel(Level.WARN)


    // 装载数据集
    val data = sc.textFile("/home/hadoop/ML_Data/input/kmeans_data.txt")
    val parsedData = data.map(s => Vectors.dense(s.split("\\s+").map(_.toDouble))).cache()

    // 将数据集聚类，4个类，50次迭代，进行模型训练形成数据模型
    val numClusters = 4
    val numIterations = 50
    val runs = 20        // 执行20次取最优
    val model = KMeans.train(parsedData, numClusters, numIterations, runs)

    //打印输出中心点坐标
    val centers = model.clusterCenters
    println("centers")
    for (i <- 0 to centers.length - 1) {
      println(centers(i)(0) + "\t" + centers(i)(1))
    }

    // 误差计算
    val WSSSE = model.computeCost(parsedData)
    println("Within Set Sum of Squared Errors = " + WSSSE)

    //给每个数据点进行标号，zippedData格式为(数据的类别号，数据)
    val c = parsedData.map(x => model.predict(x))
    c.persist()
    val zippedData = c.zip(parsedData)

    //保存zippedData到本地
    zippedData.saveAsTextFile("/home/hadoop/ML_Data/output/KMeans_data")
  }
}

3 结果展示

将结果合成到一个文件中，方便查看：