算法在ros中应用_SparkMLlib中KMeans聚类算法的解析和应用

val RANDOM &＃61; "random"
val K_MEANS_PARALLEL &＃61; "k-means||"

首先计算一个精度&＃xff1a;
val precisionBound1 &＃61; 2.0 * EPSILON * sumSquaredNorm / (normDiff * normDiff &＃43; EPSILON)
if (precisionBound1 } else if (v1.isInstanceOf[SparseVector] || v2.isInstanceOf[SparseVector]) {val dotValue &＃61; dot(v1, v2)sqDist &＃61; math.max(sumSquaredNorm - 2.0 * dotValue, 0.0)val precisionBound2 &＃61; EPSILON * (sumSquaredNorm &＃43; 2.0 * math.abs(dotValue)) /(sqDist &＃43; EPSILON)if (precisionBound2 > precision) {sqDist &＃61; Vectors.sqdist(v1, v2)}
} else {sqDist &＃61; Vectors.sqdist(v1, v2)
}
//精度不满足要求时&＃xff0c;则进行Vectors.sqdist(v1, v2)的处理&＃xff0c;即原始的距离计算

def train(// 数据样本data: RDD[Vector],// 聚类数量k: Int,// 最大迭代次数maxIterations: Int,// 初始化中心&＃xff0c;支持"random"或者"k-means||"initializationMode: String,// 初始化时的随机种子seed: Long): KMeansModel &＃61; {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 &＃61; {val sc &＃61; data.sparkContextval initStartTime &＃61; System.nanoTime()val distanceMeasureInstance &＃61; DistanceMeasure.decodeFromString(this.distanceMeasure)val centers &＃61; initialModel match {case Some(kMeansCenters) &＃61;>kMeansCenters.clusterCenters.map(new VectorWithNorm(_))case None &＃61;>if (initializationMode &＃61;&＃61; KMeans.RANDOM) {// randominitRandom(data)} else {// k-means||initKMeansParallel(data, distanceMeasureInstance)}}val initTimeInSeconds &＃61; (System.nanoTime() - initStartTime) / 1e9logInfo(f"Initialization with $initializationMode took $initTimeInSeconds%.3f seconds.")var converged &＃61; falsevar cost &＃61; 0.0var iteration &＃61; 0val iterationStartTime &＃61; System.nanoTime()instr.foreach(_.logNumFeatures(centers.head.vector.size))// Execute iterations of Lloyd&＃39;s algorithm until converged// Kmeans迭代执行&＃xff0c;计算每个样本属于哪个中心点&＃xff0c;中心点累加的样本值以及计数。然后根据中心点的所有样本数据进行中心点的更新&＃xff0c;并且比较更新前的数值&＃xff0c;根据两者距离判断是否完成//迭代次数小于最大迭代次数&＃xff0c;并行计算的中心点还没有收敛while (iteration // 当前聚类中心val thisCenters &＃61; bcCenters.value// 中心点的维度val dims &＃61; thisCenters.head.vector.sizeval sums &＃61; Array.fill(thisCenters.length)(Vectors.zeros(dims))val counts &＃61; Array.fill(thisCenters.length)(0L)points.foreach { point &＃61;>// 通过当前的聚类中心点&＃xff0c;找出最近的聚类中心点// findClosest是为了计算bestDistance&＃xff0c;参考上述Spark对距离计算的优化val (bestCenter, cost) &＃61; distanceMeasureInstance.findClosest(thisCenters, point)costAccum.add(cost)distanceMeasureInstance.updateClusterSum(point, sums(bestCenter))counts(bestCenter) &＃43;&＃61; 1}counts.indices.filter(counts(_) > 0).map(j &＃61;> (j, (sums(j), counts(j)))).iterator}.reduceByKey { case ((sum1, count1), (sum2, count2)) &＃61;>axpy(1.0, sum2, sum1)(sum1, count1 &＃43; count2)}.collectAsMap()if (iteration &＃61;&＃61; 0) {instr.foreach(_.logNumExamples(collected.values.map(_._2).sum))}val newCenters &＃61; collected.mapValues { case (sum, count) &＃61;>distanceMeasureInstance.centroid(sum, count)}bcCenters.destroy(blocking &＃61; false)// Update the cluster centers and costsconverged &＃61; truenewCenters.foreach { case (j, newCenter) &＃61;>if (converged &&!distanceMeasureInstance.isCenterConverged(centers(j), newCenter, epsilon)) {// 距离大于&＃xff0c;则说明中心点位置改变converged &＃61; false}// 更新中心点centers(j) &＃61; newCenter}cost &＃61; costAccum.valueiteration &＃43;&＃61; 1}val iterationTimeInSeconds &＃61; (System.nanoTime() - iterationStartTime) / 1e9logInfo(f"Iterations took $iterationTimeInSeconds%.3f seconds.")if (iteration &＃61;&＃61; 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)}

诺丹姆吉本斯主教中学(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
...

//将加载的rdd数据&＃xff0c;每一条变成一个向量&＃xff0c;整个数据集变成矩阵
val parsedata &＃61; rdd.map { case Row(schoolid, schoolname, locationid, school_type, zs, fee, byj) &＃61;>//"特征因子":学校位置id,学校类型,住宿方式,学费,备用金val features &＃61; Array[Double](locationid.toString.toDouble, school_type.toString.toDouble, zs.toString.toDouble, fee.toString.toDouble, byj.toString.toDouble)//将数组变成机器学习中的向量Vectors.dense(features)}.cache() //默认缓存到内存中&＃xff0c;可以调用persist()指定缓存到哪//用kmeans对样本向量进行训练得到模型//聚类中心val numclusters &＃61; List(3, 6, 9)//指定最大迭代次数val numIters &＃61; List(10, 15, 20)var bestModel: Option[KMeansModel] &＃61; Nonevar bestCluster &＃61; 0var bestIter &＃61; 0val bestRmse &＃61; Double.MaxValuefor (c <- numclusters; i <- numIters) {val model &＃61; KMeans.train(parsedata, c, i)//集内均方差总和(WSSSE)&＃xff0c;一般可以通过增加类簇的个数 k 来减小误差&＃xff0c;一般越小越好&＃xff08;有可能出现过拟合&＃xff09;val d &＃61; model.computeCost(parsedata)println("选择K:" &＃43; (c, i, d))if (d //提取到每一行的特征值val features &＃61; Array[Double](locationid.toString.toDouble, school_type.toString.toDouble, zs.toString.toDouble, fee.toString.toDouble, byj.toString.toDouble)//将特征值转换成特征向量val linevector &＃61; Vectors.dense(features)//将向量输入model中进行预测&＃xff0c;得到预测值val prediction &＃61; bestModel.get.predict(linevector)//返回每一行结果((sid,sname),所属类别)((schoolid.toString, schoolname.toString), prediction)}//中心点/*val centers: Array[linalg.Vector] &＃61; model.clusterCenterscenters.foreach(println)*///按照所属"类别"分组&＃xff0c;并根据"类别"排序&＃xff0c;然后保存到数据库// saveData2Mysql是封装好的保存数据到mysql的方法resrdd.groupBy(_._2).sortBy(_._1).foreachPartition(saveData2Mysql(_))