根据 Array() 内容对 RDD 进行排序答案

【问题标题】：Sort RDD according to an Array() content根据 Array() 内容对 RDD 进行排序
【发布时间】：2017-07-04 06:08:59
【问题描述】：

我有 RDD[(Int , Array[Double])] 例如：

1, Array(2.0,5.0,6.3)
5, Array(1.0,3.3,9.5)
1, Array(5.0,4.2,3.1)
2, Array(9.6,6.3,2.3)
1, Array(8.5,2.5,1.2)
5, Array(6.0,2.4,7.8)
2, Array(7.8,9.1,4.2)

我必须收集第一列的不同值并根据该数组排列整个 RDD。

val label_array = rdd.map(_._1).collect.distinct

输出：Array(1,5,2)，现在我要根据 label_array 来排列数据。

需要的输出

1, Array(2.0,5.0,6.3)
1, Array(5.0,4.2,3.1)
1, Array(8.5,2.5,1.2)
5, Array(1.0,3.3,9.5)
5, Array(6.0,2.4,7.8)
2, Array(9.6,6.3,2.3)
2, Array(7.8,9.1,4.2)

我试过了

val ordering = (1,5,2).productIterator.toList.zipWithIndex.toMap
rdd.sortBy{case (k,v) => ordering(k)}

但是如何获得所需的输出，因为数组会有所不同（元素和大小差异）。如何根据数组格式对RDD进行排序？

【问题讨论】：

标签： arrays sorting apache-spark rdd

【解决方案1】：

只要zipWithIndex 你的label_array 就可以了

val ordering = label_array.zipWithIndex.map(x => (x._1, x._2)).toMap

你应该有你的ordering 地图

scala.collection.immutable.Map[Int,Int] = Map(1 -> 0, 5 -> 1, 2 -> 2)

【讨论】：

【解决方案2】：

更简单的方法是创建一个具有不同第一列的新 RDD 并与之前的原始列连接

下面是一个简单的例子

val rdd = spark.sparkContext.parallelize(Seq(
        (1, Array(2.0,5.0,6.3)),
        (5, Array(1.0,3.3,9.5)),
        (1, Array(5.0,4.2,3.1)),
        (2, Array(9.6,6.3,2.3)),
        (1, Array(8.5,2.5,1.2)),
        (5, Array(6.0,2.4,7.8)),
        (2, Array(7.8,9.1,4.2))
      )
    )

    val distinct = rdd.map(v => (v._1, 1))distinct()
    //(v._1, 1)this is done because you need key value to join  

    //now join distinct with previous original RDD
    distinct.join(rdd).map(v => (v._1, v._2._2))

输出：

1, Array(2.0,5.0,6.3)
1, Array(5.0,4.2,3.1)
1, Array(8.5,2.5,1.2)
5, Array(1.0,3.3,9.5)
5, Array(6.0,2.4,7.8)
2, Array(9.6,6.3,2.3)
2, Array(7.8,9.1,4.2)

【讨论】：