Spark: создать RDD[(X, X)] всех возможных комбинаций из RDD[X]

Question

Возможно ли в Spark реализовать функцию.combination из коллекций scala?

   /** Iterates over combinations.
   *
   *  @return   An Iterator which traverses the possible n-element combinations of this $coll.
   *  @example  `"abbbc".combinations(2) = Iterator(ab, ac, bb, bc)`
   */

Например, как я могу получить от RDD[X] до RDD[List[X]] или RDD[(X,X)] для комбинаций размера = 2. И давайте предположим, что все значения в RDD являются уникальными.

Answer 1

Декартово произведение и комбинации - это две разные вещи, декартово произведение создаст СДР размера rdd.size() ^ 2 и комбинации создадут СДР размера rdd.size() choose 2

val rdd = sc.parallelize(1 to 5)
val combinations = rdd.cartesian(rdd).filter{ case (a,b) => a < b }`.
combinations.collect()

Обратите внимание, что это будет работать только в том случае, если в элементах списка определен порядок, так как мы используем <, Этот работает только для выбора двух, но может быть легко расширен, если убедиться, что отношения a < b для всех а и б в последовательности

Answer 2

Это изначально поддерживается Spark RDD с cartesian преобразование.

например:

val rdd = sc.parallelize(1 to 5)
val cartesian = rdd.cartesian(rdd)
cartesian.collect

Array[(Int, Int)] = Array((1,1), (1,2), (1,3), (1,4), (1,5), 
(2,1), (2,2), (2,3), (2,4), (2,5), 
(3,1), (3,2), (3,3), (3,4), (3,5), 
(4,1), (4,2), (4,3), (4,4), (4,5), 
(5,1), (5,2), (5,3), (5,4), (5,5))

Answer 3

Как обсуждалось, cartesian даст вам ^2 элементов декартового произведения СДР с самим собой. Этот алгоритм вычисляет комбинации (n,2) RDD без необходимости сначала вычислять n^2 элементов: (используется String как тип, обобщение до типа T требует некоторого слежения за тегами класса, которые затемняют цель здесь)

Это, вероятно, менее эффективно по сравнению с декартовой фильтрацией из-за итеративной count а также take действия, которые вынуждают вычисление СДР, но более эффективно занимают место, так как вычисляют только C(n,2) = n!/(2*(n-2))! = (n*(n-1)/2) элементы вместо n^2 декартового произведения.

 import org.apache.spark.rdd._

 def combs(rdd:RDD[String]):RDD[(String,String)] = {
    val count = rdd.count
    if (rdd.count < 2) { 
        sc.makeRDD[(String,String)](Seq.empty)
    } else if (rdd.count == 2) {
        val values = rdd.collect
        sc.makeRDD[(String,String)](Seq((values(0), values(1))))
    } else {
        val elem = rdd.take(1)
        val elemRdd = sc.makeRDD(elem)
        val subtracted = rdd.subtract(elemRdd)  
        val comb = subtracted.map(e  => (elem(0),e))
        comb.union(combs(subtracted))
    } 
 }

Answer 4

Это создает все комбинации (n, 2) и работает для любого RDD, не требуя упорядочивания элементов RDD.

val rddWithIndex = rdd.zipWithIndex
rddWithIndex.cartesian(rddWithIndex).filter{case(a, b) => a._2 < b._2}.map{case(a, b) => (a._1, b._1)}

a._2 и b._2 - это индексы, а a._1 и b._1 - это элементы исходного СДР.

Пример:

Обратите внимание, что здесь на картах не определен порядок.

val m1 = Map('a' -> 1, 'b' -> 2)
val m2 = Map('c' -> 3, 'a' -> 4)
val m3 = Map('e' -> 5, 'c' -> 6, 'b' -> 7)
val rdd = sc.makeRDD(Array(m1, m2, m3))
val rddWithIndex = rdd.zipWithIndex
rddWithIndex.cartesian(rddWithIndex).filter{case(a, b) => a._2 < b._2}.map{case(a, b) => (a._1, b._1)}.collect

Выход:

Array((Map(a -> 1, b -> 2),Map(c -> 3, a -> 4)), (Map(a -> 1, b -> 2),Map(e -> 5, c -> 6, b -> 7)), (Map(c -> 3, a -> 4),Map(e -> 5, c -> 6, b -> 7)))