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

Question

Как все знают, что разделители в 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 для использования того же ключа раздела?

Answer 1

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

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

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

---

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