scala - Spark 斯卡拉 : mapPartitions in this use case

Question

我阅读了很多有关 map 和 mapPartitions 之间差异的文章。我仍然有一些疑问。事情是在阅读之后我决定在我的代码中更改 mapPartitions 的 map 函数，因为显然 mapPartitions 比 map.

我的问题是确定我的决定在以下情况下是否正确(评论显示以前的代码):

val reducedRdd = rdd.mapPartitions(partition => partition.map(r => (r.id, r)))
//val reducedRdd = rdd.map(r => (r.id, r))
.reduceByKey((r1, r2) => r1.combineElem(r2))
// .map(e => e._2)
.mapPartitions(partition => partition.map(e => e._2))

我觉得对吗？谢谢!

Answer 1

在您的情况下，mapPartitions 应该没有任何区别。

mapPartitions vs map

当我们想要为每个分区执行一些通用计算时，mapPartitions 很有用。示例 -

rdd.mapPartitions{
  partition => 
    val complicatedRowConverter = <SOME-COSTLY-COMPUTATION>
    partition.map {
      row => (row.id, complicatedRowConverter(row) )
    }
}

在上面的例子中，我们正在创建一个 complicatedRowConverter从一些昂贵的计算中得出的函数。这个功能对整个人都是一样的 RDD 分区，我们不需要一次又一次地重新创建它。做同样事情的另一种方法可以是 -

rdd.map { row => 
      val complicatedRowConverter = <SOME-COSTLY-COMPUTATION>
      (row.id, complicatedRowConverter(row) )
    }
}

这会很慢，因为我们不必要地为每一行运行这个语句 - val complicatedRowConverter = <SOME-COSTLY-COMPUTATION> .

在您的情况下，您没有对每个分区进行任何预计算或其他任何操作。在 mapPartition 中，您只需遍历每一行并将其映射到 (row.id, row) .

所以这里的 mapPartition 不会受益，你可以使用简单的 map。