Spark：获取（键，值）RDD中每个键的前K个频繁值的有效方法？

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 中出现的频率。

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

的 RDD

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

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

Answer 2

我已经能够按如下方式圆满解决这个问题。诀窍是将问题分为两部分，第一部分将键及其值组合在一起，以获取相同 k,v 出现的次数，然后将其与新的 topk 组合器一起使用以获取出现的 topk值。

case class TopKCount(topK: Int = 10) {

  //sort and trim a traversable (String, Long) tuple by _2 value of the tuple
  def topNs(xs: TraversableOnce[(String, Long)], n: Int) = {
    var ss = List[(String, Long)]()
    var min = Long.MaxValue
    var len = 0
    xs foreach { e =>
      if (len < n || e._2 > min) {
        ss = (e :: ss).sortBy((f) => f._2)
        min = ss.head._2
        len += 1
      }
      if (len > n) {
        ss = ss.tail
        min = ss.head._2
        len -= 1
      }
    }
    ss
  }

  //seed a list for each key to hold your top N's with your first record
  def createCombiner(value: (String, Long)): Seq[(String, Long)] = Seq(value)

  //add the incoming value to the accumulating top N list for the key
  def mergeValue(combined: Seq[(String, Long)], value: (String, Long)): Seq[(String, Long)] =
    topNs(combined ++ Seq((value._1, value._2)), topK)

  //merge top N lists returned from each partition into a new combined top N list
  def mergeCombiners(combined1: Seq[(String, Long)], combined2: Seq[(String, Long)]): Seq[(String, Long)] =
    topNs(combined1 ++ combined2, topK)
}

object FieldValueCount {
  //Field Value Count combiners
  def createCombiner(value: String): (Double, Long) =
    if (Try(value.toDouble).isSuccess) (value.toDouble, 1L)
    else (0.0, 1L)

  def mergeValue(combined: (Double, Long), value: String): (Double, Long) =
    if (Try(value.toDouble).isSuccess) (combined._1 + value.toDouble, combined._2 + 1L)
    else (combined._1, combined._2 + 1L)

  def mergeCombiners(combined1: (Double, Long), combined2: (Double, Long)): (Double, Long) =
    (combined1._1 + combined2._1, combined1._2 + combined2._2)
}

// Example usage. Here input is the RDD[(String, String)]
val topKCount = TopKCount(10)

input.cache()

// combine the k,v from the original input to convert it into (k, (v, count))
val combined: RDD[(String, (String, Long))] = input.map(v => (v._1 + "|" + v._2, 1L))
  .reduceByKey(_ + _).map(k => (k._1.split("\|", -1).head, (k._1.split("\|", -1).drop(1).head, k._2)))

val topKValueCount: RDD[(String, Seq[(String, Long)])] = combined.combineByKey(
  topKCount.createCombiner,
  topKCount.mergeValue,
  topKCount.mergeCombiners
)

TopKCount 已经转换为大小写 class 以便我们可以根据需要更改 k 的值。如果不需要k可变，可以做成对象

Spark：获取（键，值）RDD中每个键的前K个频繁值的有效方法？

Spark: Efficient way to get top K frequent values per key in (key, value) RDD?

scala

mapreduce

top-n

apache-spark

rdd