Повторное разделение с фиксированным минимальным количеством элементов в каждом разделе СДР с использованием Spark

Question

У меня есть СДР со следующим количеством элементов в каждом разделе (общее количество разделов val numPart = 32:

1351, 962, 537, 250, 80, 9, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 0, 15, 88, 270, 635, 1028, 1388, 1509

Чтобы увидеть предыдущий вывод, я использую это:

def countByPartition[A](anRdd: RDD[A]): RDD[Int] = anRdd.mapPartitions(iter => Iterator(iter.length))

println(countByPartition(anRdd).collect.mkString(", "))

Я бы хотел, чтобы на каждом разделе было хотя бы минимальное количество элементов, заданное val min = 5.

Я пытался выполнить anRdd.repartition(numPart) и получаю следующее:

257, 256, 256, 256, 255, 255, 254, 253, 252, 252, 252, 252, 252, 252, 252, 252, 251, 250, 249, 248, 248, 248, 248, 248, 261, 261, 260, 260, 259, 258, 258, 257

В данном случае это было идеально, потому что в каждом разделе у меня более min элементов. Но это не всегда получается одинаково, и иногда я получаю некоторые разделы со значениями, меньшими min значения.

Есть ли способ сделать то, что я хочу?

Answer 1

Это невозможно, и, как правило, нужно выбирать разбиение так, чтобы размеры были примерно равны. Разделители в Spark в основном реализуют два метода numPartitions и getPartition.Последний является функцией от одиночного ключа до номера раздела, поэтому другие элементы и, следовательно, потенциальный размер разделов на данный момент неизвестны.