/**

  * 根据单词次数排序的wordcount

  * @author Tele

  *

  */

 public class SortWordCount {

     private static SparkConf conf = new SparkConf().setMaster("local").setAppName("sortwordcount");

     private static JavaSparkContext jsc = new JavaSparkContext(conf);

     private static String path = "D:\\inputword\\result.txt";

     public static <U> void main(String[] args) {

         JavaRDD<String> rdd = jsc.textFile(path);

         /*

          * JavaRDD<String> lines = rdd.flatMap(new FlatMapFunction<String,String>() {

          *

          * private static final long serialVersionUID = 1L;

          *

          * @Override public Iterator<String> call(String t) throws Exception { return

          * Arrays.asList(t.split(" ")).iterator(); } });

          *

          * JavaPairRDD<String, Integer> tuples = lines.mapToPair(new

          * PairFunction<String,String,Integer>() {

          *

          * private static final long serialVersionUID = 1L;

          *

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

          * return new Tuple2<String,Integer>(t,1); } });

          */

         JavaPairRDD<String, Integer> tuples = rdd.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {

             private static final long serialVersionUID = 1L;

             @Override

             public Iterator<Tuple2<String, Integer>> call(String t) throws Exception {

                 Stream<Tuple2<String, Integer>> stream = Arrays.asList(t.split(" ")).stream()

                         .map(i -> new Tuple2<>(i, 1));

                 return stream.iterator();

             }

         });

         JavaPairRDD<String, Integer> wc = tuples.reduceByKey(new Function2<Integer, Integer, Integer>() {

             private static final long serialVersionUID = 1L;

             @Override

             public Integer call(Integer v1, Integer v2) throws Exception {

                 return v1 + v2;

             }

         });

         // 将词频与单词互换位置

         JavaPairRDD<Integer, String> cw = wc.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {

             private static final long serialVersionUID = 1L;

             @Override

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

                 return new Tuple2<Integer, String>(t._2, t._1);

             }

         });

         JavaPairRDD<Integer, String> result = cw.sortByKey(false);

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

             private static final long serialVersionUID = 1L;

             @Override

             public void call(Tuple2<Integer, String> t) throws Exception {

                 System.out.println(t._2 + "----" + t._1);

             }

         });

         // 也可以在排序完毕后换成单词-词频的形式

         /*

          * JavaPairRDD<String, Integer> result = cw.sortByKey(false).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); } });

          *

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

          *

          * private static final long serialVersionUID = 1L;

          *

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

          * System.out.println(t._1 + "-------" + t._2); } });

          */

         jsc.close();

     }

 }

 object SortWordCount {

   def main(args: Array[String]): Unit = {

     val conf = new SparkConf().setMaster("local").setAppName("sortwordcount");

     val sc = new SparkContext(conf);

     val rdd = sc.textFile("D:\\inputword\\result.txt", 1);

     val wordcount = rdd.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _);

     wordcount.map(t => (t._2, t._1)).sortByKey(false, 1).map(t => (t._2, t._1)).foreach(t => println(t._1 + "-----" + t._2));

   }

 }