performance - `rdd.map(x => f(g(x))` 是否比 `rdd.map(g).map(f)` 具有更好的性能？

Question

在spark中，我们有两种操作RDD的方式。

一种是让它尽可能短:

rdd.map(x => h(f(g(x))))

另一个是链以使其更具可读性，例如:

rdd.map(g).map(f).map(h)...

个人比较喜欢后一种，比较清晰。但是有些人担心性能，他们认为它与:

list.map(g).map(f).map(h)

并认为链中会有一些即时的临时RDD，所以他们总是使用前一个。

真的吗？使用链一有任何性能问题吗？我个人把它当作 Stream而且我认为两者没有太大的性能差异

Answer 1

这些与将被流水线化的代码几乎相同。

第一个很明显，你看起来很清楚会发生什么，但是链接将导致以下结果(简化):

MapPartitionsRDD(
  MapPartitionsRDD(
    MapPartitionsRDD(
      rdd, 
      iter.map(g)), 
    iter.map(f)),
  iter.map(h))

进一步简化可视化:

map(map(map(rdd,g),f),h)

执行时归结为:

h(f(g(rddItem)))

似曾相识？它只是一个流水线计算链......通过懒惰评估的乐趣带给您。

你可以通过一个例子看到这一点:

def f(x: Int) = {println(s"f$x");x}
def g(x: Int) = {println(s"g$x");x}
def h(x: Int) = {println(s"h$x");x}

val rdd = sc.makeRDD(1 to 3, 1)

rdd.map(x => h(f(g(x))))
g1
f1
h1
g2
f2
h2
g3
f3
h3

rdd.map(g).map(f).map(h)
g1
f1
h1
g2
f2
h2
g3
f3
h3