spark 分组取topn

java

/** 
 *分组取topn,有序数列去除一些项后,仍然有序,所以应当先排序后分组
 *@author Tele
 *
 */
public class TopDemo2 {
    private static SparkConf conf = new SparkConf().setMaster("local").setAppName("topdemo2");
    private static JavaSparkContext jsc = new JavaSparkContext(conf);

    public static <U> void main(String[] args) {
        JavaRDD<String> rdd = jsc.textFile("./src/main/java/base_demo/top/score.txt");

        JavaPairRDD<Integer, String> mapToPair = rdd.mapToPair(new PairFunction<String, Integer, String>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<Integer, String> call(String t) throws Exception {
                String[] fields = t.split(" ");
                return new Tuple2<Integer, String>(Integer.parseInt(fields[1]), fields[0]);
            }
        });

        // 先排序
        JavaPairRDD<Integer, String> sortByKey = mapToPair.sortByKey(false);

        // 互换位置以便分组
        JavaPairRDD<String, Integer> mapToPair2 = sortByKey
                .mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, Integer> call(Tuple2<Integer, String> t) throws Exception {
                        return new Tuple2<String, Integer>(t._2, t._1);
                    }
                });

        // 分组
        JavaPairRDD<String, Iterable<Integer>> groupByKey2 = mapToPair2.groupByKey();

        // 取前三
        JavaPairRDD<String, Iterable<Integer>> result = groupByKey2
                .mapToPair(new PairFunction<Tuple2<String, Iterable<Integer>>, String, Iterable<Integer>>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, Iterable<Integer>> call(Tuple2<String, Iterable<Integer>> t)
                            throws Exception {

                        return new Tuple2<String, Iterable<Integer>>(t._1,
                                IteratorUtils.toList(t._2.iterator()).subList(0, 3));
                    }
                });

        result.foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() {

            private static final long serialVersionUID = 1L;

            @Override
            public void call(Tuple2<String, Iterable<Integer>> t) throws Exception {
                System.out.println(t._1 + t._2);
            }
        });

        jsc.close();
    }
}

scala

object TopDemo2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("topdemo2");
    val sc = new SparkContext(conf);

    val rdd = sc.textFile("./src/main/scala/spark_core/top/score.txt", 1);

    rdd.map(lines => {
      val fields = lines.split(" ");
      (fields(1).toInt, fields(0));
    }).sortByKey(false, 1).map(t => (t._2, t._1)).groupByKey().map(t => {
      val arr = t._2;
      val score = arr.take(3);
      (t._1, score)
    }).foreach(t => println(t._1 + "---" + t._2));
  }
}