为什么在DataFrame.rdd.map上使用DataFrame.select(反之亦然)？

Question

使用selecta DataFrame来获取我们需要的信息和为了相同的目的映射底层RDD的每一行之间是否存在"机械"差异？

"机械"我指的是执行操作的机制.换句话说,实施细节.

提供两个哪个更好/更高效？

df = # create dataframe ...
df.select("col1", "col2", ...)

要么

df = # create dataframe ...
df.rdd.map(lambda row: (row[0], row[1], ...))

我正处于性能测试的中间,因此我将找出哪个更快但我想知道什么是实现差异和优点/缺点.

Answer 1

在这个过于简单化的例子中，DataFrame.select我DataFrame.rdd.map认为差异可能几乎可以忽略不计。

毕竟您已经加载了数据集并且只进行了投影。最终，两者都必须从 Spark 的InternalRow列格式中反序列化数据，以计算操作的结果。

DataFrame.select您可以通过explain(extended = true)了解物理计划（以及物理计划）来检查发生的情况。

scala> spark.version
res4: String = 2.1.0-SNAPSHOT

scala> spark.range(5).select('id).explain(extended = true)
== Parsed Logical Plan ==
'Project [unresolvedalias('id, None)]
+- Range (0, 5, step=1, splits=Some(4))

== Analyzed Logical Plan ==
id: bigint
Project [id#17L]
+- Range (0, 5, step=1, splits=Some(4))

== Optimized Logical Plan ==
Range (0, 5, step=1, splits=Some(4))

== Physical Plan ==
*Range (0, 5, step=1, splits=Some(4))

将实际计划（即SparkPlan）与您正在做的事情rdd.map（按toDebugString）进行比较，您就会知道什么可能“更好”。

scala> spark.range(5).rdd.toDebugString
res5: String =
(4) MapPartitionsRDD[8] at rdd at <console>:24 []
 |  MapPartitionsRDD[7] at rdd at <console>:24 []
 |  MapPartitionsRDD[6] at rdd at <console>:24 []
 |  MapPartitionsRDD[5] at rdd at <console>:24 []
 |  ParallelCollectionRDD[4] at rdd at <console>:24 []

（在这个人为的例子中，我再次认为没有赢家——两者都尽可能高效）。

请注意，DataFrame它实际上Dataset[Row]用于RowEncoder将数据编码（即序列化）为InternalRow柱状二进制格式。Dataset如果要在管道中执行更多运算符，那么RDD坚持使用低级幕后逻辑查询计划优化和列式二进制格式，可以获得更好的性能。

有很多优化，试图击败它们通常会导致浪费你的时间。您必须熟记 Spark 的内部结构才能获得更好的性能（而且代价肯定是可读性）。

其中有很多内容，我强烈建议观看 Herman van Hovell 的演讲《A Deep Dive into the Catalyst Optimizer》，以了解并欣赏所有优化。

我的看法是…… “远离 RDD，除非你知道自己在做什么”。