RDD API与UDF混合DataFrame API的性能影响

Question

（Scala特定问题。）

虽然Spark docs鼓励尽可能使用DataFrame API，但如果DataFrame API不够，则通常会选择使用RDD API还是使用UDF。这两个备选方案之间是否存在固有的性能差异？

RDD和UDF的相似之处在于，它们都不能从催化剂和钨优化中获益。是否有其他开销，如果有，这两种方法之间是否有区别？

举一个具体的例子，假设我有一个DataFrame，其中包含一列具有自定义格式的文本数据（不适合正则表达式匹配）。我需要解析该列并添加一个包含结果标记的新向量列。

Answer 1

（注意：我没有为此测量背衬）

对我来说，洗牌和（反）序列化是主要成本。但在这些之后，拥有干净的代码是最重要的。记住这一点：

使用RDD操作的主要缺点是需要将/序列化为完整的jvm对象。虽然使用udf可能只会（取消）序列化所需的列。请注意，这是在处理面向列的数据时，例如拼花，对于其他我不知道的数据格式，但预计在许多情况下两者都具有相似的perf。

因此，如果您的算法主要是过滤和混洗op，并且/或者可以简单地用dataframe op和local udf表示，您应该使用它们。然而，如果您的算法需要对许多列进行复杂的处理，那么最好预先支付反序列化费用，并在jvm对象上执行干净高效的scala代码。

因此，在我实现复杂数学算法的个人经验中，我通常将代码分为两个步骤：

1. 纯dataframe op可以执行尽可能多的过滤、连接和分组操作。在极少数情况下，当需要特定的本地op时，我可以使用udf，而该op不能使用dataframe方法来表示（如果它只需要很少的列）

Answer 2

它们都不能从催化剂和钨优化中受益

这并不完全正确。虽然UDF没有从钨优化中受益（可以说简单的SQL转换也没有得到巨大的提升），但您仍然可以从Catalyst提供的执行计划优化中受益。让我们用一个简单的例子来说明这一点（注意：Spark 2.0和Scala。不要将此推断到早期版本，尤其是使用PySpark）：

val f = udf((x: String) => x == "a")
val g = udf((x: Int) => x + 1)

val df = Seq(("a", 1), ("b", 2)).toDF

df
  .groupBy($"_1")
  .agg(sum($"_2").as("_2"))
  .where(f($"_1"))
  .withColumn("_2", g($"_2"))
  .select($"_1")
  .explain

// == Physical Plan ==
// *HashAggregate(keys=[_1#2], functions=[])
// +- Exchange hashpartitioning(_1#2, 200)
//    +- *HashAggregate(keys=[_1#2], functions=[])
//       +- *Project [_1#2]
//          +- *Filter UDF(_1#2)
//             +- LocalTableScan [_1#2, _2#3]

执行计划向我们展示了几件事：

• 在聚合之前已下推所选内容
• 在聚合之前，投影被向下推，并有效地删除了第二个UDF调用

根据数据和管道，这几乎可以免费提供实质性的性能提升。

也就是说，RDD和UDF都需要在安全和不安全之间进行迁移，而后者的灵活性要低得多。尽管如此，如果您只需要一个简单的类似映射的行为，而不需要初始化昂贵的对象（如数据库连接），那么UDF就是一种选择。

在稍微复杂的场景中，您可以轻松下拉到通用Dataset并为确实需要访问自定义分区等一些低级功能的情况保留RDD。