Java에서 Spark로 값을 기준으로 요소를 정렬하는 방법

스파크는 값이 아닌 키별로 정렬 만 허용합니다. 값을 기준으로 정렬하면 키를 값으로 사용하고 값을 키로 사용하여 새 쌍에 매핑 할 수 있습니다. 그런 다음 키별로 정렬하고 출력을 렌더링합니다.

예:

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");

이 Java 예제는 Stats0 클래스를 사용하는데, 이는 정수를 감싸는 래퍼입니다. 이것은 일반 Comparator를 사용하도록 (듯이) 커스터마이즈 할 수 있습니다.이 방법에서는, 직렬화 가능한 한, 임의의 Object 형을 사용합니다.

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);
	}
}

참고 문헌 :

sparkjava