Sortieren und Ranking in Apache Spark Scala?

Question

Ich wollte in Funken tun Ranking wie folgt:

Eingang:

5.6 
5.6 
5.6 
6.2 
8.1 
5.5 
5.5 

Ranks:

1 
1 
1 
2 
3 
0 
0 
0 

Ausgang:

Rank Input 
0  5.5 
0  5.5 
1  5.6 
1  5.6 
1  5.6 
2  6.2 
3  8.1 

Ich wollte wissen, wie ich diese in Spark sortieren kann und auch die gleiche Rangfolge wie oben aufgeführt bekommen. Die Anforderungen sind:

1. Ranking beginnt mit 0 nicht 1
2. dies für Millionen von Datensätzen eine Probe Fall ist, und eine Partition kann sehr groß sein - ich Empfehlung schätzen, wie Verfahren unter Verwendung eines internen Sortier Rang

Ich wollte dies in Scala tun. Kann mir jemand dabei helfen Code zu schreiben?

Answer 1

Wenn Sie nur einige Reihen erwarten könnten Sie zunächst alle distinct Werte erhalten, sammeln sie als List und verwandeln sie in eine BroadCast. Im Folgenden zeige ich ein schmutziges Beispiel feststellen, dass es nicht, dass die Ausgabe sortiert werden garantiert wird (es wahrscheinlich Ansätze besser sein könnte, aber dies ist das erste, was mir in den Sinn kommt):

// Case 1. k is small (fits in the driver and nodes) 
val rdd = sc.parallelize(List(1,1,44,4,1,33,44,1,2)) 
val distincts = rdd.distinct.collect.sortBy(x => x) 
val broadcast = sc.broadcast(distincts) 

val sdd = rdd.map{ 
    case i: Int => (broadcast.value.asInstanceOf[Array[Int]].indexOf(i), i) 
} 

sdd.collect() 

// Array[(Int, Int)] = Array((0,1), (0,1), (4,44), (2,4), (0,1), (3,33), (4,44), (0,1), (1,2)) 

In der zweite Ansatz sortiere ich mit Spark-Funktionalität, in der RDD's documentation können Sie finden, wie zipWithIndex und keyBy arbeiten.

//case 2. k is big, distinct values don't fit in the Driver. 
val rdd = sc.parallelize(List(1,1,44,4,1,33,44,1,2)) 
val distincts = rdd.distinct.sortBy(x => x).zipWithIndex 
rdd.keyBy(x => x) 
    .join(distincts.keyBy(_._1)) 
    .map{ 
    case (value: Int, (v1: Int, (v2: Int, index: Long))) => (index, value) 
    }.collect() 

//res15: Array[(Long, Int)] = Array((3,33), (2,4), (0,1), (0,1), (0,1), (0,1), (4,44), (4,44), (1,2)) 

By the way, verwende ich collect nur zur Visualisierung, in einer realen Anwendung, die Sie es nicht, wenn Sie sicher sind, verwenden sollte es im Fahrer Speicher passt.