在 apache Spark 中替换 groupByKey 的方法

我想知道更换的最佳方法按键分组与另一个操作。

基本上我想获得一个RDD[(int,列表[测量])， 我的情况：

// consider measures like RDD of objects
measures.keyBy(_.getId)
        .groupByKey

我的想法是使用按键减少相反，因为它会导致更少的随机播放：

measures.keyBy(_.getId)
        .mapValues(List(_))
        .reduceByKey(_++_) 

但我认为效率非常低，因为它迫使我实例化大量不必要的 List 对象。

任何人都可以有其他想法来替换 groupByKey 吗？

另一种方法是使用aggregateByKey，专门用于将值组合成与原始值不同的类型：

measures.keyBy(_.getId)
        .aggregateByKey(List[Measure]())(_ :+ _, _ ++ _)

这会为每个分区中的每个键创建一个空列表，将所有值附加到每个分区中的这些值，然后最后打乱列表以连接每个键的所有值。

在 Scala 中向列表追加是 O(n)，最好是在前面添加，即 O(1)，但看起来不太干净：

measures.keyBy(_.getId)
        .aggregateByKey(List[Measure]())(_.+:(_), _ ++ _)

or:

measures.keyBy(_.getId)
        .aggregateByKey(List[Measure]())((l, v) => v +: l, _ ++ _)

这可能比你的更有效率reduceByKey示例，但情况reduceByKey and aggregateByKey远远优于groupByKey您可以首先大幅减少数据大小，然后仅对较小的结果进行打乱。在这种情况下，您没有这种减少：中间列表包含您开始使用的所有数据，因此当组合每个分区列表时，您仍在对完整数据集进行洗牌（这对于使用reduceByKey).

此外，正如 Zero323 指出的那样，groupByKey在这种情况下实际上更有效，因为它知道正在构建所有数据的列表，并且可以专门为此执行优化：

• 它禁用映射端聚合，从而阻止使用所有数据构建大哈希映射
• 它使用智能缓冲区（CompactBuffer），与逐一构建不可变列表相比，这显着减少了内存分配量。

另一种情况是两者之间的差异groupByKey and reduceByKey or aggregateByKey当键的数量不比值的数量少很多时，可能是最小的。