草庐IT

关于scala：Spark：在(键，值)RDD中获取每个键的前K个频繁值的有效方法？

codeneng 2023-03-28 原文

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

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

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
    }
  }

我是这样使用的：

1
2
3
4
5
6
// 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 条目发送到驱动程序，而不是发送整个 Map，我在当前情况下正在这样做。

非常感谢任何反馈。

  • 不确定这是否可行，而不是使用地图，创建自定义地图。欺骗此自定义 Map 以仅序列化 TopK 对象。

我已经能够令人满意地解决这个问题，如下所示。诀窍是将问题分成两部分，在第一部分将键及其值组合在一起，以获取相同 k，v 发生的次数，然后将其与新的 topk 组合器一起使用以获取发生的 topk价值观。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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 已转换为案例类，以便我们可以根据需要更改 k 的值。如果 k 不需要是可变的，它可以作为一个对象。

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

1
2
//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

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

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

  • groupBy 有效，但效率不高。原因是洗牌太多，所有数据都传递给驱动程序来完成这项工作。 combineByKey 领先一步，您可以在其中指定如何在分区级别进行分组，然后将此分组数据发送到驱动程序以与来自其他分区的数据合并
  • @sushant-hiray 对不起，你是对的，combinedByKey 似乎更有效率。

频繁scalaspanclassbrapache-sparkmapreducerddtop-n

有关关于scala：Spark：在(键，值)RDD中获取每个键的前K个频繁值的有效方法？的更多相关文章

  1. ruby - 如何使用 Nokogiri 的 xpath 和 at_xpath 方法 - 2

    我正在学习如何使用Nokogiri，根据这段代码我遇到了一些问题:require'rubygems'require'mechanize'post_agent=WWW::Mechanize.newpost_page=post_agent.get('http://www.vbulletin.org/forum/showthread.php?t=230708')puts"\nabsolutepathwithtbodygivesnil"putspost_page.parser.xpath('/html/body/div/div/div/div/div/table/tbody/tr/td/div

  2. ruby - 如何从 ruby​​ 中的字符串运行任意对象方法？ - 2

    总的来说，我对ruby​​还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用

  3. ruby - 为什么我可以在 Ruby 中使用 Object#send 访问私有(private)/ protected 方法？ - 2

    类classAprivatedeffooputs:fooendpublicdefbarputs:barendprivatedefzimputs:zimendprotecteddefdibputs:dibendendA的实例a=A.new测试a.foorescueputs:faila.barrescueputs:faila.zimrescueputs:faila.dibrescueputs:faila.gazrescueputs:fail测试输出failbarfailfailfail.发送测试[:foo,:bar,:zim,:dib,:gaz].each{|m|a.send(m)resc

  4. ruby - Facter::Util::Uptime:Module 的未定义方法 get_uptime (NoMethodError) - 2

    我正在尝试设置一个puppet节点，但ruby​​gems似乎不正常。如果我通过它自己的二进制文件(/usr/lib/ruby/gems/1.8/gems/facter-1.5.8/bin/facter)在cli上运行facter，它工作正常，但如果我通过由ruby​​gems(/usr/bin/facter)安装的二进制文件，它抛出:/usr/lib/ruby/1.8/facter/uptime.rb:11:undefinedmethod`get_uptime'forFacter::Util::Uptime:Module(NoMethodError)from/usr/lib/ruby

  5. Ruby 方法() 方法 - 2

    我想了解Ruby方法methods()是如何工作的。我尝试使用“ruby方法”在Google上搜索，但这不是我需要的。我也看过ruby​​-doc.org，但我没有找到这种方法。你能详细解释一下它是如何工作的或者给我一个链接吗？更新我用methods()方法做了实验，得到了这样的结果:'labrat'代码classFirstdeffirst_instance_mymethodenddefself.first_class_mymethodendendclassSecond使用类#returnsavailablemethodslistforclassandancestorsputsSeco

  6. ruby-on-rails - Rails 3.2.1 中 ActionMailer 中的未定义方法 'default_content_type=' - 2

    我在我的项目中添加了一个系统来重置用户密码并通过电子邮件将密码发送给他，以防他忘记密码。昨天它运行良好(当我实现它时)。当我今天尝试启动服务器时，出现以下错误。=>BootingWEBrick=>Rails3.2.1applicationstartingindevelopmentonhttp://0.0.0.0:3000=>Callwith-dtodetach=>Ctrl-CtoshutdownserverExiting/Users/vinayshenoy/.rvm/gems/ruby-1.9.3-p0/gems/actionmailer-3.2.1/lib/action_mailer

  7. ruby - Highline 询问方法不会使用同一行 - 2

    设置:狂欢ruby1.9.2高线(1.6.13)描述:我已经相当习惯在其他一些项目中使用highline，但已经有几个月没有使用它了。现在，在Ruby1.9.2上全新安装时，它似乎不允许在同一行回答提示。所以以前我会看到类似的东西:require"highline/import"ask"Whatisyourfavoritecolor?"并得到:Whatisyourfavoritecolor?|现在我看到类似的东西:Whatisyourfavoritecolor?|竖线(|)符号是我的终端光标。知道为什么会发生这种变化吗？ 最佳答案

  8. ruby - 主要 :Object when running build from sublime 的未定义方法 `require_relative' - 2

    我已经从我的命令行中获得了一切，所以我可以运行rubymyfile并且它可以正常工作。但是当我尝试从sublime中运行它时，我得到了undefinedmethod`require_relative'formain:Object有人知道我的sublime设置中缺少什么吗？我正在使用OSX并安装了rvm。 最佳答案 或者，您可以只使用“require”，它应该可以正常工作。我认为“require_relative”仅适用于ruby​​1.9+ 关于ruby-主要:Objectwhenrun

  9. ruby - 多个属性的 update_column 方法 - 2

    我有一个具有一些属性的模型:attr1、attr2和attr3。我需要在不执行回调和验证的情况下更新此属性。我找到了update_column方法，但我想同时更新三个属性。我需要这样的东西:update_columns({attr1:val1,attr2:val2,attr3:val3})代替update_column(attr1,val1)update_column(attr2,val2)update_column(attr3,val3) 最佳答案 您可以使用update_columns(attr1:val1,attr2:val2

  10. ruby - 检查方法参数的类型 - 2

    我不确定传递给方法的对象的类型是否正确。我可能会将一个字符串传递给一个只能处理整数的函数。某种运行时保证怎么样？我看不到比以下更好的选择:defsomeFixNumMangler(input)raise"wrongtype:integerrequired"unlessinput.class==FixNumother_stuffend有更好的选择吗？ 最佳答案 使用Kernel#Integer在使用之前转换输入的方法。当无法以任何合理的方式将输入转换为整数时，它将引发ArgumentError。defmy_method(number)

随机推荐