Схема разбиения по умолчанию в Spark

Question

Когда я выполняю нижеприведенную команду:

scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)),4).partitionBy(new HashPartitioner(10)).persist()
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[10] at partitionBy at <console>:22

scala> rdd.partitions.size
res9: Int = 10

scala> rdd.partitioner.isDefined
res10: Boolean = true


scala> rdd.partitioner.get
res11: org.apache.spark.Partitioner = org.apache.spark.HashPartitioner@a

Это говорит о том, что есть 10 разделов, и разделение сделано, используя HashPartitioner, Но когда я выполню нижеприведенную команду:

scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)),4)
...
scala> rdd.partitions.size
res6: Int = 4
scala> rdd.partitioner.isDefined
res8: Boolean = false

Там написано, что есть 4 раздела и разделитель не определен. Итак, что такое схема секционирования по умолчанию в Spark? / Как данные разбиты во втором случае?

Answer 1

Вы должны различать две разные вещи:

• разделение как распределение данных между разделами в зависимости от значения ключа, которое ограничено только PairwiseRDDs (RDD[(T, U)]). Это создает связь между разделом и набором ключей, которые можно найти в данном разделе.

• разбиение как разделение ввода на несколько разделов, где данные просто делятся на порции, содержащие последовательные записи, чтобы обеспечить распределенные вычисления. Точная логика зависит от конкретного источника, но это либо количество записей, либо размер фрагмента.

  В случае parallelize данные равномерно распределяются между разделами с использованием индексов. В случае HadoopInputFormats (лайк textFile) это зависит от свойств, таких как mapreduce.input.fileinputformat.split.minsize / mapreduce.input.fileinputformat.split.maxsize,

Таким образом, схема секционирования по умолчанию просто отсутствует, потому что секционирование не применимо ко всем СДР. Для операций, которые требуют разбиения на PairwiseRDD (aggregateByKey, reduceByKey и т. д.) по умолчанию используется разделение хешей.