[数据挖掘] - 聚类算法：K-means算法理解及SparkCore实现

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer
import scala.util.Random

/**
  * SparkCore实现KMeans算法
  * Created by gerry
  */
object SparkCoreKMeansDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("spark-core-kmeans")
      .setMaster("local[*]")
    val sc = SparkContext.getOrCreate(conf)
    //=======================================
    val path = "data/k-means/result"
    val fs = FileSystem.get(sc.hadoopConfiguration)
    fs.delete(new Path(path), true)
    val rdd = sc.textFile("data/taxi/taxi.csv").map(line => {
      val arr = line.split(",")
      (arr(1).toDouble, arr(2).toDouble)
    })

    rdd.cache()

    // SparkCore实现K-Means
    val K = 10
    // 获取初始化种子
    var clusters = rdd.takeSample(false, K)
    var totalDistance: Double = 0.0

    do {
      // 将k个聚类中心广播出去
      val clusterPointWithIndex = clusters.zipWithIndex.map(_.swap)
      val seeder = sc.broadcast(clusterPointWithIndex)

      // 开始计算新的聚类中心
      val newClusterInfos = rdd
        .map(point => {
          // 计算当前节点离所有节点最近的节点是那个
          val clusterIndex = seeder.value
            .map {
              case (index, clusterPoint) => {
                val dist = calcDistance(point, clusterPoint)
                (dist, index)
              }
            }
            .minBy(_._1)._2

          // 返回结果
          (clusterIndex, point)
        })
        .aggregateByKey(ClusterPoint(0D, 0D, 1L))(
          (clusterPoint, point) => {
            // 累加
            clusterPoint.x += point._1
            clusterPoint.y += point._2
            clusterPoint.n += 1
            // 返回聚合结果
            clusterPoint
          },
          (clusterPoint1, clusterPoint2) => {
            // 累加
            clusterPoint1.x += clusterPoint2.x
            clusterPoint1.y += clusterPoint2.y
            clusterPoint1.n += clusterPoint2.n

            // 返回聚合结果
            clusterPoint1
          }
        )
        .collect()
        .map {
          case (oldClusterIndex, clusterPoint) => {
            // 计算新的中心点
            val newCluster = (clusterPoint.x / clusterPoint.n, clusterPoint.y / clusterPoint.n)
            // 获取上一个迭代产生的聚簇点
            val oldClusterPoint = clusterPointWithIndex
              .filter(_._1 == oldClusterPoint)
              .toList match {
              case (_, point) :: Nil => point
              case _ => throw new RuntimeException("不应该出现该异常，从之前的聚簇集合中没法找到下标为")
            }

            // 计算老的中心点和新的中心点的距离
            val distance = calcDistance(oldClusterPoint, newCluster)
            // 返回新的节点, 以及距离
            (distance, newCluster)
          }
        }

      // 清空广播变量
      seeder.unpersist(true)

      // 获得总距离(两次C集合之间的变化距离)
      totalDistance = newClusterInfos.map(_._1).sum
      // 新的聚类点集合
      clusters = newClusterInfos.map(_._2)
      println("distance:" + totalDistance + "," + clusters.mkString("{", ",", "}"))
    } while (totalDistance > 1e-7)

    rdd.map(tuple => tuple._1 + "\t" + tuple._2).repartition(1).saveAsTextFile(path)
    println("最终聚簇点:" + clusters.map(_.swap).mkString("{", ",", "}"))

    rdd.unpersist()
  }

  /**
    * 集群间点的表示方式
    *
    * @param x
    * @param y
    * @param n
    */
  case class ClusterPoint(var x: Double, var y: Double, var n: Long) extends Serializable

  /**
    * 计算距离(欧几里得距离)
    *
    * @param point1
    * @param point2
    * @return
    */
  def calcDistance(point1: (Double, Double), point2: (Double, Double)): Double = {
    import scala.math._
    val a = point1._1 - point2._1
    val b = point1._2 - point2._2
    return sqrt(pow(a, 2) + pow(b, 2))
  }
}