scala - Spark : Efficient way to get top K frequent values per key in (key, 值)RDD？

Question

我有一个(键，值)对的 RDD。我需要根据每个键的频率获取前 k 个值。

我知道最好的方法是使用 combineByKey。

目前这是我的 combineByKey 组合器的样子

object TopKCount {
    //TopK Count combiners
    val k: Int = 10
    def createCombiner(value: String): Map[String, Long] = {
      Map(value -> 1L)
    }
    def mergeValue(combined: Map[String, Long], value: String): Map[String, Long] = {
      combined ++ Map(value -> (combined.getOrElse(value, 0L) + 1L))
    }
    def mergeCombiners(combined1: Map[String, Long], combined2: Map[String, Long]): Map[String, Long] = {
      val top10Keys1 = combined1.toList.sortBy(_._2).takeRight(k).toMap.keys
      val top10Keys2 = combined2.toList.sortBy(_._2).takeRight(k).toMap.keys

      (top10Keys1 ++ top10Keys2).map(key => (key, combined1.getOrElse(key, 0L) + combined2.getOrElse(key, 0L)))
        .toList.sortBy(_._2).takeRight(k).toMap
    }
  }

我使用它如下:

// input is RDD[(String, String)]
 val topKValueCount: RDD[(String, Map[String, Long])] = input.combineByKey(
      TopKCount.createCombiner,
      TopKCount.mergeValue,
      TopKCount.mergeCombiners
    )

对当前代码的一个优化是在 mergeCombiners 期间使用 min-queue。

我更关心网络 I/O。是否有可能一旦我在一个分区中进行合并，我只将该分区中的 topK 条目发送到驱动程序，而不是发送整个映射，这是我在当前情况下所做的。

非常感谢任何反馈。

Answer 1

为什么不使用 Spark 的 RDD GroupByKey 功能或 GroupBy？如果您正在处理大型 RDD，使用 Spark 功能几乎总是更快，对吗？

//assuming input is RDD[(String, String)]
val groupinput = input.groupBy(_._2).map(x=>(x._1,x._2.map(y=>y._2).groupBy(identity).map(z=>(z._1,z._2.size)).toList.sortBy(-_._2)))

这条紧凑的 1 行应该做你想做的。该行首先按键对 RDD 进行分组，输出 RDD(keys, Map(Key,values))。现在，第二个 GroupBy 对 Mapping 的值进行分组，并输出这些值在新 Map 中出现的频率。

最后，我将 map 转换为列表(使用数组或您认为合适的任何内容)并按计数(或频率)排序。所以你有一个

的 RDD

RDD[(key, List[(value, frequency)])]

现在您可以在 List 上使用 take(k) 来获取 k 个最频繁的值。