Spark solo permite ordenar por claves y no por valores. La ordenación por valores se puede lograr mapeando a un nuevo par con la clave como el valor y el valor como la clave. Luego ordenando por clave y renderizando la salida.
long start = System.nanoTime();
JavaPairRDD<String, Stats0> extracted = dataSet1.mapToPair(s -> new Tuple2<>(getIp(s), new Stats0(1)));
JavaPairRDD<String, Stats0> baseKeyPair = extracted.reduceByKey(Stats0::merge);
// Map for sorting
JavaPairRDD<Integer, Tuple2<String, Stats0>> sortingRDD = baseKeyPair
.mapToPair(t ->new Tuple2<>(t._2().getCount(), t));
// Sort by keys
sortingRDD = sortingRDD.sortByKey(false);
// Collect to display the output
List<Tuple2<Integer, Tuple2<String, Stats0>>> output = sortingRDD.collect();
end = System.nanoTime();
for (Tuple2<Integer, Tuple2<String, Stats0>> t : output) {
System.out.println(t._2()._1 + "\t" + t._1());
}
System.out.println("Processed in : " + (int) (end - start)/1000000 + " ms");
Este exemplo Java está usando a classe Stats0, que é um invólucro em torno de um inteiro. Isso pode ser personalizado para usar um comparador genérico, e dessa forma use qualquer tipo de objeto, desde que seja serializável.
public static class Stats0 implements Serializable {
private final int count;
public Stats0(int count) {
this.count = count;
}
public Stats0 merge(Stats0 other) {
return new Stats0(count + other.count);
}
public int getCount() {
return count;
}
public String toString() {
return String.format("n=%s", count);
}
}