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

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

 
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 = [email protected]

В нем указано, что существует 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?/Как данные разделяются во втором случае?

Ответы

Ответ 1

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

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

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

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

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