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作为 a 的结果flatMap（并将结果转换回 a Dataset）。对混乱表示歉意。我还添加了有关提高分析性能的更多信息。

更新 2：我错过了您使用的是一个Dataset而不是一个RDD（doh！）。这不会显着影响答案。

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

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

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

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