Spark MLlib中KMeans聚类算法的解析和应用

简介: 聚类算法是机器学习中的一种无监督学习算法,它在数据科学领域应用场景很广泛,比如基于用户购买行为、兴趣等来构建推荐系统。

本文转自公众号:大数据学习与分享
原文链接


聚类算法是机器学习中的一种无监督学习算法,它在数据科学领域应用场景很广泛,比如基于用户购买行为、兴趣等来构建推荐系统。

核心思想可以理解为,在给定的数据集中(数据集中的每个元素有可被观察的n个属性),使用聚类算法将数据集划分为k个子集,并且要求每个子集内部的元素之间的差异度尽可能低,而不同子集元素的差异度尽可能高。简而言之,就是通过聚类算法处理给定的数据集,将具有相同或类似的属性(特征)的数据划分为一组,并且不同组之间的属性相差会比较大。

K-Means算法是聚类算法中应用比较广泛的一种聚类算法,比较容易理解且易于实现。

"标准" K-Means算法

KMeans算法的基本思想是随机给定K个初始簇中心,按照最邻近原则把待分类样本点分到各个簇。然后按平均法重新计算各个簇的质心,从而确定新的簇心。一直迭代,直到簇心的移动距离小于某个给定的值或者满足已定条件。主要分为4个步骤:

1. 为要聚类的点寻找聚类中心,比如随机选择K个点作为初始聚类中心
2. 计算每个点到聚类中心的距离,将每个点划分到离该点最近的聚类中去
3. 计算每个聚类中所有点的坐标平均值,并将这个平均值作为新的聚类中心
4. 反复执行第2步和第3步,直到聚类中心不再改变或者聚类次数达到设定迭代上限或者达到指定的容错范围

示例图:

image.png

KMeans算法在做聚类分析的过程中主要有两个难题:初始聚类中心的选择和聚类个数K的选择。

Spark MLlib对KMeans的实现分析

Spark MLlib针对"标准"KMeans的问题,在实现自己的KMeans上主要做了如下核心优化:

1. 选择合适的初始中心点

Spark MLlib在初始中心点的选择上,有两种算法:

随机选择:依据给的种子seed,随机选择K个随机中心点
k-means||:默认的算法

val RANDOM = "random"
val K_MEANS_PARALLEL = "k-means||"

2. 计算样本属于哪一个中心点时对距离计算的优化

假设中心点是(a1,b1),要计算的点是(a2,b2),那么Spark MLlib采取的计算方法是(记为lowerBoundOfSqDist):

image.png

对比欧几里得距离(记为EuclideanDist):
image.png

可轻易证明lowerBoundOfSqDist小于或等于EuclideanDist,并且计算lowerBoundOfSqDist很方便,只需处理中心点和要计算的点的L2范数。那么在实际处理中就分两种情况:

  • 当lowerBoundOfSqDist大于"最近距离"(之前计算好的,记为bestdistance),那么可以推导欧式距离也大于bestdistance,不需要计算欧式距离,省去了很多计算工作
  • 当lowerBoundOfSqDist小于bestdistance,则会调用fastSquaredDistance进行距离的快速计算

关于fastSquaredDistance:

首先计算一个精度:
val precisionBound1 = 2.0 * EPSILON * sumSquaredNorm / (normDiff * normDiff + EPSILON)
if (precisionBound1 < precision) {
   // 精度满足squared distance期望的精度
   // val sumSquaredNorm = norm1 * norm1 + norm2 * norm2
   // 2.0 * dot(v1, v2)为2(a1*a2 + b1*b2)可以利用之前计算的L2范数
   sqDist = sumSquaredNorm - 2.0 * dot(v1, v2)
} else if (v1.isInstanceOf[SparseVector] || v2.isInstanceOf[SparseVector]) {
   val dotValue = dot(v1, v2)
   sqDist = math.max(sumSquaredNorm - 2.0 * dotValue, 0.0)
   val precisionBound2 = EPSILON * (sumSquaredNorm + 2.0 * math.abs(dotValue)) /
     (sqDist + EPSILON)
   if (precisionBound2 > precision) {
      sqDist = Vectors.sqdist(v1, v2)
   }
} else {
  sqDist = Vectors.sqdist(v1, v2)
}
//精度不满足要求时,则进行Vectors.sqdist(v1, v2)的处理,即原始的距离计算

Spark MLlib中KMeans相关源码分析

基于mllib包下的KMeans相关源码涉及的类和方法(ml包下与下面略有不同,比如涉及到的fit方法):

  1. KMeans类和伴生对象
  2. train方法:根据设置的KMeans聚类参数,构建KMeans聚类,并执行run方法进行训练
  3. run方法:主要调用runAlgorithm方法进行聚类中心点等的核心计算,返回KMeansModel
  4. initialModel:可以直接设置KMeansModel作为初始化聚类中心选择,也支持随机和k-means || 生成中心点
  5. predict:预测样本属于哪个"类"
  6. computeCost:通过计算数据集中所有的点到最近中心点的平方和来衡量聚类效果。一般同样的迭代次数,cost值越小,说明聚类效果越好。

注意:该方法在Spark 2.4.X版本已经过时,并且会在Spark 3.0.0被移除,具体取代方法可以查看ClusteringEvaluator

主要看一下train和runAlgorithm的核心源码:
def train(
      // 数据样本
      data: RDD[Vector],
      // 聚类数量
      k: Int,
      // 最大迭代次数
      maxIterations: Int,
      // 初始化中心,支持"random"或者"k-means||"
      initializationMode: String,
      // 初始化时的随机种子
      seed: Long): KMeansModel = {
  new KMeans().setK(k)
      .setMaxIterations(maxIterations)
      .setInitializationMode(initializationMode)
      .setSeed(seed)
      .run(data)
}
/**
   * Implementation of K-Means algorithm.
   */
  private def runAlgorithm( data: RDD[VectorWithNorm],
      instr: Option[Instrumentation]): KMeansModel = {

    val sc = data.sparkContext

    val initStartTime = System.nanoTime()

    val distanceMeasureInstance = DistanceMeasure.decodeFromString(this.distanceMeasure)

    val centers = initialModel match {
      case Some(kMeansCenters) =>
        kMeansCenters.clusterCenters.map(new VectorWithNorm(_))
      case None =>
        if (initializationMode == KMeans.RANDOM) {
          // random
          initRandom(data)
        } else {
          // k-means||
          initKMeansParallel(data, distanceMeasureInstance)
        }
    }
    val initTimeInSeconds = (System.nanoTime() - initStartTime) / 1e9
    logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")

    var converged = false
    var cost = 0.0
    var iteration = 0

    val iterationStartTime = System.nanoTime()

    instr.foreach(_.logNumFeatures(centers.head.vector.size))

    // Execute iterations of Lloyd's algorithm until converged
    // Kmeans迭代执行,计算每个样本属于哪个中心点,中心点累加的样本值以及计数。然后根据中心点的所有样本数据进行中心点的更新,并且比较更新前的数值,根据两者距离判断是否完成
    //迭代次数小于最大迭代次数,并行计算的中心点还没有收敛
    while (iteration < maxIterations && !converged) {
      // 损失值累加器
      val costAccum = sc.doubleAccumulator
      // 广播中心点
      val bcCenters = sc.broadcast(centers)

      // Find the new centers
      val collected = data.mapPartitions { points =>
        // 当前聚类中心
        val thisCenters = bcCenters.value
        // 中心点的维度
        val dims = thisCenters.head.vector.size

        val sums = Array.fill(thisCenters.length)(Vectors.zeros(dims))
        val counts = Array.fill(thisCenters.length)(0L)

        points.foreach { point =>
          // 通过当前的聚类中心点,找出最近的聚类中心点
          // findClosest是为了计算bestDistance,参考上述Spark对距离计算的优化
          val (bestCenter, cost) = distanceMeasureInstance.findClosest(thisCenters, point)
          costAccum.add(cost)
          distanceMeasureInstance.updateClusterSum(point, sums(bestCenter))
          counts(bestCenter) += 1
        }

        counts.indices.filter(counts(_) > 0).map(j => (j, (sums(j), counts(j)))).iterator
      }.reduceByKey { case ((sum1, count1), (sum2, count2)) =>
        axpy(1.0, sum2, sum1)
        (sum1, count1 + count2)
      }.collectAsMap()

      if (iteration == 0) {
        instr.foreach(_.logNumExamples(collected.values.map(_._2).sum))
      }

      val newCenters = collected.mapValues { case (sum, count) =>
        distanceMeasureInstance.centroid(sum, count)
      }

      bcCenters.destroy(blocking = false)

      // Update the cluster centers and costs
      converged = true
      newCenters.foreach { case (j, newCenter) =>
        if (converged &&
          !distanceMeasureInstance.isCenterConverged(centers(j), newCenter, epsilon)) {
          // 距离大于,则说明中心点位置改变
          converged = false
        }
        // 更新中心点
        centers(j) = newCenter
      }

      cost = costAccum.value
      iteration += 1
    }

    val iterationTimeInSeconds = (System.nanoTime() - iterationStartTime) / 1e9
    logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")

    if (iteration == maxIterations) {
      logInfo(s"KMeans reached the max number of iterations: $maxIterations.")
    } else {
      logInfo(s"KMeans converged in $iteration iterations.")
    }

    logInfo(s"The cost is $cost.")

    new KMeansModel(centers.map(_.vector), distanceMeasure, cost, iteration)
  }

Spark MLlib的KMeans应用示例

1.准备数据

诺丹姆吉本斯主教中学(Notre Dame-Bishop Gibbons School)      71     0       0    283047.0        13289.0
海景基督高中(Ocean View Christian Academy)      45     0     0       276403.0       13289.0 
卡弗里学院(Calvary Baptist Academy)        58       0       0       227567.0       13289.0
...

2.示例代码

//将加载的rdd数据,每一条变成一个向量,整个数据集变成矩阵
val parsedata = rdd.map { case Row(schoolid, schoolname, locationid, school_type, zs, fee, byj) =>
   //"特征因子":学校位置id,学校类型,住宿方式,学费,备用金
   val features = Array[Double](locationid.toString.toDouble, school_type.toString.toDouble, zs.toString.toDouble, fee.toString.toDouble, byj.toString.toDouble)
    //将数组变成机器学习中的向量
    Vectors.dense(features)
  }.cache() //默认缓存到内存中,可以调用persist()指定缓存到哪

  //用kmeans对样本向量进行训练得到模型
  //聚类中心
  val numclusters = List(3, 6, 9)
  //指定最大迭代次数
  val numIters = List(10, 15, 20)
  var bestModel: Option[KMeansModel] = None
  var bestCluster = 0
  var bestIter = 0
  val bestRmse = Double.MaxValue
  for (c <- numclusters; i <- numIters) {
    val model = KMeans.train(parsedata, c, i)
    //集内均方差总和(WSSSE),一般可以通过增加类簇的个数 k 来减小误差,一般越小越好(有可能出现过拟合)
    val d = model.computeCost(parsedata)
    println("选择K:" + (c, i, d))
    if (d < bestRmse) {
      bestModel = Some(model)
      bestCluster = c
      bestIter = i
    }
  }
  println("best:" + (bestCluster, bestIter, bestModel.get.computeCost(parsedata)))
  //用模型对我们的数据进行预测
  val resrdd = df.map { case Row(schoolid, schoolname, locationid, school_type, zs, fee, byj) =>
  //提取到每一行的特征值
  val features = Array[Double](locationid.toString.toDouble, school_type.toString.toDouble, zs.toString.toDouble, fee.toString.toDouble, byj.toString.toDouble)
   //将特征值转换成特征向量
   val linevector = Vectors.dense(features)
   //将向量输入model中进行预测,得到预测值
   val prediction = bestModel.get.predict(linevector)

   //返回每一行结果((sid,sname),所属类别)
   ((schoolid.toString, schoolname.toString), prediction)
 }

 //中心点
 /*val centers: Array[linalg.Vector] = model.clusterCenters
 centers.foreach(println)*/

 //按照所属"类别"分组,并根据"类别"排序,然后保存到数据库
 // saveData2Mysql是封装好的保存数据到mysql的方法
 resrdd.groupBy(_._2).sortBy(_._1).foreachPartition(saveData2Mysql(_))

上述示例只是一个简单的demo,实际应用中会更复杂,牵涉到数据的预处理,比如对数据进行量化、归一化,以及如何调参以获取最优训练模型。


阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区,定期推送精彩案例,技术专家直播,问答区近万人Spark技术同学在线提问答疑,只为营造纯粹的Spark氛围,欢迎钉钉扫码加入!image.png
对开源大数据和感兴趣的同学可以加小编微信(下图二维码,备注“进群”)进入技术交流微信群。image.png
Apache Spark技术交流社区公众号,微信扫一扫关注
image.png

相关文章
|
8天前
|
机器学习/深度学习 数据采集 自然语言处理
理解并应用机器学习算法:神经网络深度解析
【5月更文挑战第15天】本文深入解析了神经网络的基本原理和关键组成,包括神经元、层、权重、偏置及损失函数。介绍了神经网络在图像识别、NLP等领域的应用,并涵盖了从数据预处理、选择网络结构到训练与评估的实践流程。理解并掌握这些知识,有助于更好地运用神经网络解决实际问题。随着技术发展,神经网络未来潜力无限。
|
3天前
|
存储 Kubernetes API
使用Kubernetes管理容器化应用的深度解析
【5月更文挑战第20天】本文深度解析Kubernetes在管理容器化应用中的作用。Kubernetes是一个开源平台,用于自动化部署、扩展和管理容器,提供API对象描述应用资源并维持其期望状态。核心组件包括负责集群控制的Master节点(含API Server、Scheduler、Controller Manager和Etcd)和运行Pod的工作节点Node(含Kubelet、Kube-Proxy和容器运行时环境)。
|
2天前
|
移动开发 Java Android开发
构建高效的Android应用:内存优化策略解析
【5月更文挑战第21天】在移动开发领域,尤其是面向资源受限的Android设备,内存管理与优化是提升应用性能和用户体验的关键因素。本文深入探讨了Android内存优化的多个方面,包括内存泄漏的预防、合理的内存分配策略、以及有效的内存回收机制。通过分析内存管理的原理和提供实用的编码实践,开发者可以显著减少其应用的内存占用,从而避免常见的性能瓶颈和应用程序崩溃问题。
|
2天前
|
并行计算 Java 大数据
Java 8新特性解析及应用
【5月更文挑战第21天】本文将深入探讨Java 8的新特性,包括Lambda表达式、Stream API、Optional类以及新的日期时间API。我们将详细解析这些特性的使用方法和应用场景,帮助读者更好地理解和应用Java 8的新特性。
|
3天前
|
机器学习/深度学习 编解码 算法
算法工程师面试问题总结 | YOLOv5面试考点原理全解析
本文给大家带来的百面算法工程师是深度学习目标检测YOLOv5面试总结,文章内总结了常见的提问问题,旨在为广大学子模拟出更贴合实际的面试问答场景。在这篇文章中,我们还将介绍一些常见的深度学习目标检测面试问题,并提供参考的回答及其理论基础,以帮助求职者更好地准备面试。通过对这些问题的理解和回答,求职者可以展现出自己的深度学习目标检测领域的专业知识、解决问题的能力以及对实际应用场景的理解。同时,这也是为了帮助求职者更好地应对深度学习目标检测岗位的面试挑战,提升面试的成功率和竞争力。
|
6天前
|
存储 并行计算 算法
大模型量化技术解析和应用
眼看人工智能含智能量越来越高含人量越来越低,是否开始担心自己要跟不上这趟高速列车了?内心是否也充满好奇:大模型背后的奥秘是什么?为何如此强大?它能为我所用吗?哪种技术最适合我的需求?
|
7天前
|
缓存 算法 Java
数据结构~缓存淘汰算法--LRU算法(Java的俩种实现方式,万字解析
数据结构~缓存淘汰算法--LRU算法(Java的俩种实现方式,万字解析
|
7天前
|
缓存 算法 前端开发
前端开发者必知的缓存淘汰策略:LRU算法解析与实践
前端开发者必知的缓存淘汰策略:LRU算法解析与实践
|
8天前
|
机器学习/深度学习 算法 数据挖掘
【机器学习】聚类算法中,如何判断数据是否被“充分”地聚类,以便算法产生有意义的结果?
【5月更文挑战第14天】【机器学习】聚类算法中,如何判断数据是否被“充分”地聚类,以便算法产生有意义的结果?
|
8天前
|
算法 PyTorch Go
深入解析yolov5,为什么算法都是基于yolov5做改进的?(一)
深入解析yolov5,为什么算法都是基于yolov5做改进的?(一)

推荐镜像

更多