В Apache Spark, почему RDD.union не сохраняет разделитель?

Как все знают, что разделители в Spark оказывают огромное влияние на производительность любых "широких" операций, поэтому он обычно настраивается в операциях. Я экспериментировал со следующим кодом:

val rdd1 =
  sc.parallelize(1 to 50).keyBy(_ % 10)
    .partitionBy(new HashPartitioner(10))
val rdd2 =
  sc.parallelize(200 to 230).keyBy(_ % 13)

val cogrouped = rdd1.cogroup(rdd2)
println("cogrouped: " + cogrouped.partitioner)

val unioned = rdd1.union(rdd2)
println("union: " + unioned.partitioner)

Я вижу, что по умолчанию cogroup() всегда дает RDD с настраиваемым разделителем, но union() не работает, он всегда возвращается к умолчанию. Это нелогично, поскольку мы обычно предполагаем, что PairRDD должен использовать свой первый элемент в качестве ключа раздела. Есть ли способ "заставить" Spark объединить 2 PairRDD для использования одного и того же ключа раздела?

Ответ 1

union - очень эффективная операция, поскольку она не перемещает никаких данных. Если rdd1 имеет 10 разделов, а rdd2 имеет 20 разделов, то rdd1.union(rdd2) будет иметь 30 разделов: разделы двух RDD, вставленных друг за другом. Это просто бухгалтерское изменение, нет перетасовки.

Но обязательно он отбрасывает разделитель. Для заданного количества разделов построен разделитель. В результате RDD имеет несколько разделов, которые отличаются от rdd1 и rdd2.

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

Есть одно исключение из вышеизложенного. Если rdd1 и rdd2 имеют одинаковый разделитель (с одинаковым количеством разделов), union ведет себя по-разному. Он присоединяется к разделам двух RDD попарно, указывая на то же количество разделов, что и каждый из входов. Это может включать перемещение данных вокруг (если разделы не были расположены друг с другом), но не будет тасовать. В этом случае разделитель сохраняется. (Код для этого находится в PartitionerAwareUnionRDD.scala.)

Ответ 2

Это больше не правда. Если два RDD имеют одинаковый разделитель и количество разделов, RDD union ed также будет иметь те же разделы. Это было введено в https://github.com/apache/spark/pull/4629 и включено в Spark 1.3.