scala - flatmap 是否比 filter+map 提供更好的性能？

Question

我有一个相当大的数据集(1 亿多条记录，有 100 列)，我正在用 spark 进行处理。我正在将数据读入 spark 数据集，我想过滤此数据集并将其字段的子集映射到案例类。

代码看起来有些相似，

case class Subset(name:String,age:Int)
case class Complete(name:String,field1:String,field2....,age:Int)

val ds = spark.read.format("csv").load("data.csv").as[Complete]

#approach 1
ds.filter(x=>x.age>25).map(x=> Subset(x.name,x.age))

#approach 2
ds.flatMap(x=>if(x.age>25) Seq(Subset(x.name,x.age)) else Seq.empty)

哪种方法更好？关于如何使此代码更具性能的任何其他提示？

谢谢!

编辑

我运行了一些测试来比较运行时，看起来方法 2 更快，我用于获取运行时的代码如下，

val subset = spark.time {
   ds.filter(x=>x.age>25).map(x=> Subset(x.name,x.age))
}

spark.time {
   subset.count()
}

and 

val subset2 = spark.time {
   ds.flatMap(x=>if(x.age>25) Seq(Subset(x.name,x.age)) else Seq.empty)
}

spark.time {
   subset2.count()
}

Answer 1

更新 :我原来的回答包含一个错误:Spark 确实支持 Seq由于 flatMap (并将结果转换回 Dataset )。对困惑表示歉意。我还添加了有关提高分析性能的更多信息。

更新 2 :我错过了您使用的是 Dataset而不是 RDD (哦!)。这不会显着影响答案。

Spark 是一个分布式系统，它跨多个节点对数据进行分区并并行处理数据。在效率方面，导致重新分区(需要在节点之间传输数据)的操作在运行时方面比就地修改要昂贵得多。另外，您应该注意仅转换数据的操作，例如 filter , map , flatMap等只是存储，直到执行操作操作(例如 reduce 、 fold 、 aggregate 等)才会执行。因此，就目前情况而言，这两种选择实际上都无济于事。

当对这些转换的结果执行操作时，我希望 filter操作效率更高:它只处理通过谓词 map 的数据(使用后续 x=>x.age>25 操作) (通常写为 _.age > 25 )。虽然看起来 filter创建一个中间集合，它会延迟执行。结果，Spark 似乎融合了 filter和 map一起操作。

您的 flatMap坦率地说，操作是可怕的。它强制每个数据项的处理、序列创建和随后的扁平化，这肯定会增加整体处理。

也就是说，提高分析性能的最佳方法是控制分区，以便将数据大致均匀地拆分到尽可能多的节点上。引用 this guide作为一个很好的起点。