为什么 Spark DataFrame 转换为 RDD 需要完全重新映射？

Question

来自 Spark 源代码：

/**
   * Represents the content of the Dataset as an `RDD` of `T`.
   *
   * @group basic
   * @since 1.6.0
   */
  lazy val rdd: RDD[T] = {
    val objectType = exprEnc.deserializer.dataType
    rddQueryExecution.toRdd.mapPartitions { rows =>
      rows.map(_.get(0, objectType).asInstanceOf[T])
    }
  }

https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala#L2972

mapPartitions 可能需要与首先计算 RDD 一样长的时间。所以这使得诸如

df.rdd.getNumPartitions

非常昂贵。鉴于DataFrame 是DataSet[Row] 并且DataSet 由RDD 组成，为什么需要重新映射？任何见解都值得赞赏。

Answer 1

TL;DR 那是因为内部的RDD 不是RDD[Row]。

假设一个DataFrame是DataSet[Row]，一个DataSet是由RDD组成的

这是一个巨大的过度简化。首先DataSet[T] 并不意味着你与T 的容器交互。这意味着如果您使用类集合 API（通常称为强类型），内部表示将被解码为T。

内部表示是 Tungsten 内部使用的二进制格式。这种表示是内部的，可能会发生变化，而且级别太低，无法在实践中使用。

公开此数据的中间表示是InternalRow - rddQueryExecution.toRDD 实际上是RDD[InternalRow]。这种表示（有不同的实现）仍然公开内部类型，被认为是“弱”私有的，因为 o.a.s.sql.catalyst 中的所有对象（访问没有明确限制，但 API 没有记录），并且交互起来相当棘手。

这就是解码发挥作用的地方，以及为什么您需要完全“重新映射” - 将通常不安全的内部对象转换为供公众使用的外部类型。

最后，to reiterate 我之前的语句 - 当调用 getNumPartitions 时，相关代码不会被执行。