使用 sortBy 对 JavaRDD 进行排序

Question

我有一个JavaRDD<Tuple2<String, Integer>>，想根据它的 Integer 字段对其进行排序。

如果RDD只有一个字段，它可以正常工作。

JavaRDD<Integer> sorted = wordSet.sortBy(x -> x, false, 10);

但是由于它的字段是 Tuple2，所以它不起作用。

我尝试了以下代码：

JavaRDD<Tuple2<String, Integer>> sorted1 = wordSet1.sortBy(new Function<Tuple2<String, Integer>, Integer>() {
        public Integer call(Tuple2<String, Integer> value) throws Exception {
            return value._2;
        }
    }, false, 10);

我也尝试使用Comparator，但它也不起作用。

public class TupleComparator implements Comparator<Tuple2<String, Integer>>, Serializable
{
    @Override
    public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
        return o1._2() < o2._2() ? 0 : 1;
    }
}

任何帮助将不胜感激。

更新：一个完整​​且可验证的示例：

public class sortRDD {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("sort");
        JavaSparkContext sparkContext = new JavaSparkContext(conf);

        List<Integer> numbers = new ArrayList<Integer>();
        for (int i = 1; i < 10; i++) {
            numbers.add(i);
        }
        JavaRDD<Integer> numberRDD = sparkContext.parallelize(numbers);
        System.out.println(numberRDD.collect());
        JavaRDD<Integer> sorted = numberRDD.sortBy(x -> x, false, 1);
        System.out.println("sorted:");
        System.out.println(sorted.collect());

        //RDD of tuples
        List<Tuple2<String, Integer>> words = new ArrayList<Tuple2<String, Integer>>();
        for (int i = 1; i < 10; i++) {
            words.add(new Tuple2<String, Integer>("word" + i, i));
        }
        JavaRDD<Tuple2<String, Integer>> wordSet = sparkContext.parallelize(words);
        System.out.println(wordSet.collect());
        JavaRDD<Tuple2<String, Integer>> sorted1 = wordSet.sortBy(new Function<Tuple2<String, Integer>, Integer>() {
            public Integer call(Tuple2<String, Integer> value) throws Exception {
                return value._2;
            }
        }, false, 1);
        System.out.println("sorted:");
        System.out.println(sorted1.collect());

        //Using Comparator
        JavaRDD<Tuple2<String, Integer>> sorted2 = wordSet.sortBy(new TupleComparator(), false, 1);
        System.out.println("sorted:");
        System.out.println(sorted1.collect());
    }
}

Answer 1

我将JavaRDD改为JavaPairRdd，然后使用sortByKey，我可以得到排序后的结果。