Как проверить количество разделов Spark DataFrame без затрат на .rdd

Нет внутренней стоимости rdd компонента в rdd.getNumPartitions, потому что возвращенный RDD никогда не оценивается.

Хотя вы можете легко определить это эмпирически, используя отладчик (я оставлю это в качестве упражнения для читателя) или установив, что никакие задания не запускаются в базовом сценарии

Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.0
      /_/

Using Scala version 2.11.12 (OpenJDK 64-Bit Server VM, Java 1.8.0_181)
Type in expressions to have them evaluated.
Type :help for more information.

scala> val ds = spark.read.text("README.md")
ds: org.apache.spark.sql.DataFrame = [value: string]

scala> ds.rdd.getNumPartitions
res0: Int = 1

scala> spark.sparkContext.statusTracker.getJobIdsForGroup(null).isEmpty // Check if there are any known jobs
res1: Boolean = true

этого может быть недостаточно, чтобы убедить вас. Итак, давайте подойдем к этому более систематично:

• rdd возвращает MapPartitionRDD (ds, как определено выше):

  scala> ds.rdd.getClass
  res2: Class[_ <: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]] = class org.apache.spark.rdd.MapPartitionsRDD
  

• RDD.getNumPartitions RDD.partitions.

• В сценарии без контрольной точки _ 12_ вызывает getPartitions (также не стесняйтесь отслеживать путь контрольной точки).
• RDD.getPartitions .
• Таким образом, фактическая реализация, используемая в этом случае, - MapPartitionsRDD.getPartitions, что просто делегирует вызов родителю.

• Между rdd и источником есть только MapPartitionsRDD.

  scala> ds.rdd.toDebugString
  res3: String =
  (1) MapPartitionsRDD[3] at rdd at <console>:26 []
   |  MapPartitionsRDD[2] at rdd at <console>:26 []
   |  MapPartitionsRDD[1] at rdd at <console>:26 []
   |  FileScanRDD[0] at rdd at <console>:26 []
  

  Точно так же, если Dataset содержит обмен, мы будем следовать за родителями до ближайшего тасования:

  scala> ds.orderBy("value").rdd.toDebugString
  res4: String =
  (67) MapPartitionsRDD[13] at rdd at <console>:26 []
   |   MapPartitionsRDD[12] at rdd at <console>:26 []
   |   MapPartitionsRDD[11] at rdd at <console>:26 []
   |   ShuffledRowRDD[10] at rdd at <console>:26 []
   +-(1) MapPartitionsRDD[9] at rdd at <console>:26 []
      |  MapPartitionsRDD[5] at rdd at <console>:26 []
      |  FileScanRDD[4] at rdd at <console>:26 []
  

  Обратите внимание, что этот случай особенно интересен, потому что мы фактически запустили задание:

  scala> spark.sparkContext.statusTracker.getJobIdsForGroup(null).isEmpty
  res5: Boolean = false
  
  scala> spark.sparkContext.statusTracker.getJobIdsForGroup(null)
  res6: Array[Int] = Array(0)
  

  Это потому, что мы столкнулись со сценарием, когда разделы не могут быть определены статически (см. Количество разделов фрейма данных после сортировки? и Почему преобразование sortBy запускает задание Spark?).

  В таком сценарии getNumPartitions также вызовет задание:

  scala> ds.orderBy("value").rdd.getNumPartitions
  res7: Int = 67
  
  scala> spark.sparkContext.statusTracker.getJobIdsForGroup(null)  // Note new job id
  res8: Array[Int] = Array(1, 0)
  

  однако это не означает, что наблюдаемая стоимость каким-то образом связана с .rdd звонком. Вместо этого это внутренняя стоимость поиска partitions в случае, когда нет статической формулы (например, некоторые входные форматы Hadoop, где требуется полное сканирование данных).

Обратите внимание, что приведенные здесь замечания не следует экстраполировать на другие приложения Dataset.rdd. Например, ds.rdd.count было бы действительно дорого и расточительно.

По моему опыту df.rdd.getNumPartitions работает очень быстро, я никогда не сталкивался с тем, что это занимает больше секунды или около того.

В качестве альтернативы вы также можете попробовать

val numPartitions: Long = df
      .select(org.apache.spark.sql.functions.spark_partition_id()).distinct().count()

что позволило бы избежать использования .rdd