是否将同一个RDD上的多个减速器编译成单个扫描？

Question

假设我有一个RDD（50M记录/day redu），我想用几种不同的方式来总结。RDD记录是4元组：（保持、foo、bar、baz）。

• 保持-布尔
• foo，条形，baz-0/1整数

我想数一下每个< code>foo有多少个

rdd.filter(lambda keep, foo, bar, baz: foo == 1)
   .map(lambda keep, foo, bar, baz: keep, 1)
   .reduceByKey(operator.add)

这将返回（在收集之后）一个列表，如[（True，40000000），（False，10000000）]。

问题是:有没有一种简单的方法可以避免扫描< code>rdd三次(对< code>foo 、< code>bar 、< code>baz各扫描一次)？

我的意思不是重写上面的代码来处理所有3个字段的方法，而是告诉Spark在一次传递中处理所有3个管道。

Answer 1

如果我没看错你的问题你想要RDD.aggregate.

val zeroValue = (0L, 0L, 0L, 0L, 0L, 0L) // tfoo, tbar, tbaz, ffoo, fbar, fbaz
rdd.aggregate(zeroValue)(
  (prior, current) => if (current._1) {
    (prior._1 + current._2, prior._2 + current._3, prior._3 + current._4,
      prior._4, prior._5, prior._6)
  } else {
    (prior._1, prior._2, prior._3,
      prior._4 + current._2, prior._5 + current._3, prior._6 + current._4)
  },
  (left, right) =>
    (left._1 + right._1,
      left._2 + right._2,
      left._3 + right._3,
      left._4 + right._4,
      left._5 + right._5,
      left._6 + right._6)
)

聚合在概念上类似于列表上的概念化简函数，但 RDD 不是列表，它们是分布式的，因此您提供了两个函数参数，一个用于对每个分区进行操作，另一个用于组合处理分区的结果。

Answer 2

通过使用不同的线程提交作业，可以并行执行三条管道，但这将通过RDD三次，并且需要集群上最多3倍的资源。

通过重写作业以一次处理所有计数，可以一次性完成工作 - 关于聚合的答案是一个选项。将数据成对拆分（keep，foo）（keep，bar），（keep，baz）将是另一个。

不修改任何代码，一次完成任务是不可能的，因为Spark无法知道这些任务与同一个数据集相关。在第一个作业之后，最多可以通过在< code >之前使用< code>rdd.cache对初始rdd进行< code >缓存来提高后续作业的速度。过滤器()。地图()。reduce()步骤；这仍将通过RDD 3次，但如果所有数据都适合群集的内存，第二次和第三次可能会快得多:

rdd.cache
// first reduceByKey action will trigger the cache and rdd data will be kept in memory
val foo = rdd.filter(fooFilter).map(fooMap).reduceByKey(???)
// subsequent operations will execute faster as the rdd is now available in mem
val bar = rdd.filter(barFilter).map(barMap).reduceByKey(???)
val baz = rdd.filter(bazFilter).map(bazMap).reduceByKey(???)

若我这样做，我会创建相关数据对，并在一次通过中对它们进行计数：

// We split the initial tuple into pairs keyed by the data type ("foo", "bar", "baz") and the keep information. dataPairs will contain data like: (("bar",true),1), (("foo",false),1)

val dataPairs = rdd.flatmap{case (keep, foo, bar, baz) => 
    def condPair(name:String, x:Int):Option[((String,Boolean), Int)] = if (x==1) Some(((name,keep),x)) else None
    Seq(condPair("foo",foo), condPair("bar",bar), condPair("baz",baz)).flatten
}
val totals = dataPairs.reduceByKey(_ + _)

这很简单，只会传递一次数据，但需要重写代码。我认为它在回答这个问题时得分为66,66%。