统计一下每一个月份中，温度最高top2

思路
首先按照月份来分组，对组内的数据按照温度来排序
取温度最高的前两名，然后分组取RDD
代码

public class TopNTemperature {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("TopNTemperature").setMaster("local[1]");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> weatherRdd = sc.textFile("weather");
        weatherRdd = weatherRdd.cache();
        JavaPairRDD<String, List<Double>> top2TemperaturePerMonthRDD = weatherRdd.mapToPair(new PairFunction<String, String,Double>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, Double> call(String log) throws Exception {
                String[] splited = log.split("\t");
                String date = splited[0].substring(0, 7);
                Double temperature = Double.parseDouble(splited[1].trim());
                return new Tuple2<String, Double>(date,temperature);
            }
        })//生成（年月，温度）的RDD
        .groupByKey()//按照月份分组
        .mapToPair(new PairFunction<Tuple2<String,Iterable<Double>>, String,List<Double>>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, List<Double>> call(Tuple2<String, Iterable<Double>> tuple) throws Exception {
                String date = tuple._1;
                Iterator<Double> iterator = tuple._2.iterator();
                Double[] temperatures = new Double[2];
                //list
                
                while(iterator.hasNext()){
                    //遍历每一个月份
                    Double temperature = iterator.next();
                    for(int i = 0 ;i < temperatures.length ; i++){
                        //遍历每一个月份里的温度
                        if(temperatures[i] == null){
                            //判断某一个月份里是否只有一个温度，如果有不用排序了
                            temperatures[i] = temperature;
                            break;
                        }else{
                            if(temperature > temperatures[i]){
                                /** 
                                 * 有两个以上的温度，对Value值排序，取出最大的两个数，类似于冒泡法
                                 * temperature[]这个数组只能装下两个元素多于的就被淘汰了
                                 */
                                
                                
                                for(int j = (temperatures.length-1) ; j > i ;j--){
                                    temperatures[j] = temperatures[j-1];
                                }
                                temperatures[i] = temperature;
                                break;
                            }
                        }
                    }
                }
                //Collections.sort()
                
                List<Double> asList = Arrays.asList(temperatures);
                
                return new Tuple2<String, List<Double>>(date,asList);
            }
        });
        
        List<Tuple2<String, List<Double>>> collect = top2TemperaturePerMonthRDD.collect();
        for(Tuple2<String,List<Double>> t:collect){
            //collect算子：Action算子，能够将每一个task的计算结果回收到Driver端，一般用于测试
            String date = t._1;
            List<Double> topN = t._2;
            System.out.println("date:" + date + "\tTemperature:" + topN.get(0) + "\t" + topN.get(1));
        }
        
        sc.stop();
    }
}

关于serialVersionUID
serialVersionUID适用于Java的序列化机制。简单来说，Java的序列化机制是通过判断类的serialVersionUID来验证版本一致性的。在进行反序列化时，JVM会把传来的字节流中的serialVersionUID与本地相应实体类的serialVersionUID进行比较，如果相同就认为是一致的，可以进行反序列化，否则就会出现序列化版本不一致的异常，即是InvalidCastException。