GroupByKey 比 CombineByKey 快 [重复]

Question

我在 RDD 中有 (Int, (Int, Int, Int) 类型的元素。目的是将具有相同键的元素限制为某个阈值t。更简单的解决方案如下：

rdd.groupByKey().flatMapValues{iterable => {
  iterable.take(t)
}}

我认为将这段代码替换为 combineByKey 会很有用，以便通过组合器利用地图端聚合，因为可能有超过 t 单个分区中的相同键，导致以下结果：

val function_createCombiner = (x: (Int, Int, Int)) => {
   ArrayBuffer[(Int, Int, Int)](x)
 }
val function_mergeValue = (accumulator: ArrayBuffer[(Int, Int, Int)],
                           x: (Int, Int, Int)) => {
  if(accumulator.size < t){
    accumulator += x
  }
  accumulator
}
val function_mergeCombiners = (accumulator1: ArrayBuffer[(Int, Int, Int)],
                               accumulator2: ArrayBuffer[(Int, Int, Int)]) => {
  val iter = accumulator2.iterator
  var saturated = false
  while(!saturated && iter.hasNext){
    if(accumulator1.length < t){
      accumulator1 += iter.next()
    } else {
      saturated = true
    }
  }
  accumulator1
}

rdd
.combineByKey(function_createCombiner, function_mergeValue, function_mergeCombiners)
.flatMapValues(x => x.toList)

令人惊讶的是，combineByKey 解决方案的性能比 groupByKey 解决方案差。 combineByKey 解决方案的 GC 工作时间为 50%，所以我认为我创建了许多临时缓冲区。另一方面，网上到处都在说，应该尽量避免 groupByKey。

CombineByKey 时间：11 分钟

GroupByKey 时间：4.1 分钟

我的 combineByKey 解决方案是否存在一些可怕的缺陷？还是我错过了什么？

提前致谢！

编辑：这个问题实际上是重复的，我很抱歉。这是因为只有非常少量的元素出现超过 t 次。因此很明显，我（几乎）尝试通过 combineByKey 重新实现 groupByKey。唯一的选择是我使用 groupByKey，这似乎更快，或者如果可能的话完全省略该步骤。 无论如何，感谢您的帮助！

Answer 1

就个人而言，与rdds合作时，我会选择reduceByKey:)

rdd
 .mapValues(List(_))
 .reduceByKey((v1, v2) => (v1++v2).take(t))
 .flatMapValues(identity(_))

我发现它比combineByKey 容易得多，而且它通常比groupByKey 更有效，因为它在重新排列数据之前确实映射了边减少。我说通常是因为在某些情况下（例如，当您收集每个键的所有值时）groupByKey 的性能与reduceByKey 一样好。