用reduceByKey（）替换groupByKey（）

Question

这是这里的后续问题。我正试图在这个实现的基础上实现k-means。它工作得很好，但我想用reduceByKey（）替换groupByKey（），但我不确定如何替换（我现在不担心性能）。以下是相关的简化代码：

val data = sc.textFile("dense.txt").map(
        t => (t.split("#")(0), parseVector(t.split("#")(1)))).cache()

val read_mean_centroids = sc.textFile("centroids.txt").map(
        t => (t.split("#")(0), parseVector(t.split("#")(1))))
var centroids = read_mean_centroids.takeSample(false, K, 42).map(x => x._2)
do {
    var closest = read_mean_centroids.map(p => (closestPoint(p._2, centroids), p._2))
    var pointsGroup = closest.groupByKey() // <-- THE VICTIM :)
    var newCentroids = pointsGroup.mapValues(ps => average(ps.toSeq)).collectAsMap()
    ..

请注意，println（newCentroids）将给出：

地图（23）-

和println（最接近的）：

在地图kmeans.scala:75

相关问题：在ApacheSpark（Scala）中使用reduceByKey。

一些文件：

DEF还原ByKey（func：（V， V）0020 V）： RDD[（K， V）]

使用关联减少函数合并每个关键点的值。

DEF减少ByKey（func：（V， V）0020 V， Num分区： Int）： RDD[（K， V）]

使用关联减少函数合并每个关键点的值。

def reduceByKey（partitioner:partitioner，func:（V，V）⇒ 五） ：RDD[（K，V）]

使用关联减少函数合并每个关键点的值。

def groupByKey（）：RDD[（K，Iterable[V]）]

将RDD中每个键的值分组为单个序列。

Answer 1

你可以像这样使用一个aggregateByKey（）（比reduceebykey（）自然一点）来计算新的质心：

val newCentroids = closest.aggregateByKey((Vector.zeros(dim), 0L))(
  (agg, v) => (agg._1 += v, agg._2 + 1L),
  (agg1, agg2) => (agg1._1 += agg2._1, agg1._2 + agg2._2)
).mapValues(agg => agg._1/agg._2).collectAsMap 

为此，您需要计算数据的维度，即dim，但只需计算一次。您可能会使用类似于val dim=data的东西。首先_2.长度。