Почему для преобразования 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 не задокументирован), и довольно сложно взаимодействовать.

Это когда декодирование вступает в игру и поэтому вам необходимо полное «переопределение» - для преобразования внутренних, часто небезопасных, объектов во внешние типы, предназначенные для публичного использования.

Наконец, для повторения моего предыдущего утверждения - рассматриваемый код не будет выполнен при вызове getNumPartitions.