Spark - сортировка значений Double в СДР и игнорирование NaN - PullRequest
Новая
       23

Spark - сортировка значений Double в СДР и игнорирование NaN

0 голосов
/ 15 ноября 2018

Я хочу отсортировать значения Double в СДР и хочу, чтобы моя функция сортировки игнорировала значения Double.NaN.

Либо значения Double.NaN должны отображаться в нижней или верхней части отсортированного СДР.

Мне не удалось этого добиться с помощью sortBy. 

scala> res13.sortBy(r => r, ascending = true)
res21: org.apache.spark.rdd.RDD[Double] = MapPartitionsRDD[10] at sortBy at <console>:26

scala> res21.collect.foreach(println)
0.656
0.99
0.998
1.0
NaN
5.6
7.0

scala> res13.sortBy(r => r, ascending = false)
res23: org.apache.spark.rdd.RDD[Double] = MapPartitionsRDD[15] at sortBy at <console>:26

scala> res23.collect.foreach(println)
7.0
5.6
NaN
1.0
0.998
0.99
0.656

Myожидаемый результат 

scala> res23.collect.foreach(println)
    7.0
    5.6
    1.0
    0.998
    0.99
    0.656
    NaN

or 
    scala> res21.collect.foreach(println)
    NaN
    0.656
    0.99
    0.998
    1.0
    5.6
    7.0

Ответы [ 2 ]

0 голосов
/ 16 ноября 2018

Чтобы добавить ответ @ user3685285: 

scala> def sortAscending(r: Double): Double = { if (r.isNaN) Double.MaxValue else r }
sortAscending: (r: Double)Double

scala> def sortDescending(r: Double): Double = {if (r.isNaN) Double.MinValue else r }
sortDescending: (r: Double)Double

scala> res0.sortBy(sortDescending, ascending=false)
res7: org.apache.spark.rdd.RDD[Double] = MapPartitionsRDD[20] at sortBy at <console>:28

scala> res7.collect.foreach(println)
99.9
34.2
10.98
7.0
6.0
5.0
2.0
0.56
0.01
0.0
NaN
NaN

scala> res0.sortBy(sortAscending, ascending=true)
res9: org.apache.spark.rdd.RDD[Double] = MapPartitionsRDD[25] at sortBy at <console>:28

scala> res9.collect.foreach(println)
0.0
0.01
0.56
2.0
5.0
6.0
7.0
10.98
34.2
99.9
NaN
NaN
0 голосов
/ 16 ноября 2018

Взяв то, что я сказал в комментарии, вы можете попробовать это: 

scala> val a = sc.parallelize(Array(0.656, 0.99, 0.998, 1.0, Double.NaN, 5.6, 7.0))
a: org.apache.spark.rdd.RDD[Double] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> a.sortBy(r => r, ascending = false).collect
res2: Array[Double] = Array(7.0, 5.6, NaN, 1.0, 0.998, 0.99, 0.656)

scala> a.sortBy(r => if (r.isNaN) Double.MinValue else r, ascending = false).collect
res3: Array[Double] = Array(7.0, 5.6, 1.0, 0.998, 0.99, 0.656, NaN)

scala> a.sortBy(r => if (r.isNaN) Double.MaxValue else r, ascending = false).collect
res4: Array[Double] = Array(NaN, 7.0, 5.6, 1.0, 0.998, 0.99, 0.656)
...