1.定义输入和输出为:

• 输入数据格式为：年，月，日，温度。格式为2012，01，01，05
• 输出数据的格式为：年-月,温度。 格式为：2012-01，3，30 ，35。

2.使用MapReduce来完成上述二次排序的需求

- 2.1使用到的类的介绍

• SecondarySortDrivcer 驱动器类，定义输入输出，并注册插件类
• SecondarySortMapper 定义map()函数
• SecondarySortReduce 定义reduce()函数
• DataTemperatureGroupingComparator 定义如何对键分组
• DataTemperaturePair 将日期和温度对定义为java对象
• DataTemperaturePartitioner 定义定制分区器

- 2.2 关于启动类SecondarySortDriver.java

//SecondarySortDriver.java

public static int submitJob(String[] args) throws Exception {
        //inputDir和outputDir存入为HDFS路径的话，则会连接到HDFS
        //String[] args = new String[2];
        //args[0] = inputDir;
        //args[1] = outputDir;
        //ToolRunner.run 的方法中实现了tool中的run，上面的对其进行了重写
        int returnStatus = ToolRunner.run(new SecondarySortDriver(), args);
        return returnStatus;
    }
    
//ToolRunner.run方法，其中tool.run方法是接口Tool的方法

 public static int run(Configuration conf, Tool tool, String[] args) throws Exception {
        if(conf == null) {
            conf = new Configuration();
        }

        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        tool.setConf(conf);
        String[] toolArgs = parser.getRemainingArgs();
        return tool.run(toolArgs);
    }

//之后就重写的run方法了

     @Override
    public int run(String[] args) throws Exception {
        //定义了一些job的提交方式
        Configuration conf = getConf();
        Job job = new Job(conf);
        job.setJarByClass(SecondarySortDriver.class);
        job.setJobName("SecondarySortDriver");
        
            // args[0] = input directory
            // args[1] = output directory
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //定义job的提交方式
        job.setOutputKeyClass(DateTemperaturePair.class);
        job.setOutputValueClass(Text.class);
        
        job.setMapperClass(SecondarySortMapper.class);
        job.setReducerClass(SecondarySortReducer.class);        
        job.setPartitionerClass(DateTemperaturePartitioner.class);
        job.setGroupingComparatorClass(DateTemperatureGroupingComparator.class);

        boolean status = job.waitForCompletion(true);
        theLogger.info("run(): status="+status);
        return status ? 0 : 1;
    }

- 2.3 关于Map函数类SecondarySortMapper.java

  private final Text theTemperature = new Text();
    //DateTemperaturePair可以理解为DateTemperaturePair，day,temperature的实体类，提供比较方法
    private final DateTemperaturePair pair = new DateTemperaturePair();

 
    protected void map(LongWritable key, Text value, Context context) 
        throws IOException, InterruptedException {
        String line = value.toString();
        String[] tokens = line.split(",");
        // YYYY = tokens[0]
        // MM = tokens[1]
        // DD = tokens[2]
        // temperature = tokens[3]
        String yearMonth = tokens[0] + tokens[1];
        String day = tokens[2];
        int temperature = Integer.parseInt(tokens[3]);

        pair.setYearMonth(yearMonth);
        pair.setDay(day);
        pair.setTemperature(temperature);
        theTemperature.set(tokens[3]);

        context.write(pair, theTemperature);
    }
    

• 其中DateTemperaturePair的比较方法为：

    //如果yearMonth相同的话，才进行排序
    @Override
    public int compareTo(DateTemperaturePair pair) {
        int compareValue = this.yearMonth.compareTo(pair.getYearMonth());
        if (compareValue == 0) {
            compareValue = temperature.compareTo(pair.getTemperature());
        }
        //return compareValue;      // to sort ascending 
        return -1*compareValue;     // to sort descending 
    }

- 2.4 关于Map函数类SecondarySortReducer.java

//注意传入的参数DateTemperaturePair, Text, Text, Text
public class SecondarySortReducer 
    extends Reducer<DateTemperaturePair, Text, Text, Text> {

    @Override
    protected void reduce(DateTemperaturePair key, Iterable<Text> values, Context context) 
        throws IOException, InterruptedException {
        StringBuilder builder = new StringBuilder();
        for (Text value : values) {
            builder.append(value.toString());
            builder.append(",");
        }
        context.write(key.getYearMonth(), new Text(builder.toString()));
    }
}

- 2.5 关于运行函数的脚本

#cat run.sh

export JAVA_HOME=/usr/java/jdk8
export BOOK_HOME=/home/mp/data-algorithms-book
export APP_JAR=$BOOK_HOME/dist/data_algorithms_book.jar
INPUT=/secondary_sort/input
OUTPUT=/secondary_sort/output
$HADOOP_HOME/bin/hadoop -fs -rmr $OUTPUT
PROG = org.dataalgorithms.chap01.mapreduce.SecondarySortDriver
$HADOOP_HOME/bin/hadoop jar $APP_JAR $PROG $INPUT $OUTPUT

3.使用spark来完成上述二次排序的需求

- 3.1 需求的定义

 * Input:
 *
 *    name, time, value  
 *    x,2,9
 *    y,2,5
 *    x,1,3
 *    y,1,7
 *    y,3,1
 *    x,3,6
 *    z,1,4
 *    z,2,8
 *    z,3,7
 *    z,4,0
 *
 *Output: generate a time-series looking like this:
 *  x => [(1,3), (2,9), (3,6)]
 *  y => [(1,7), (2,5), (3,1)]
 *  z => [(1,4), (2,8), (3,7), (4,0)]

- 3.2 SecondarySort类总结构

public class SecondarySort{
     public static void main(String[] args) throws Exception {
        //步骤2：读取输入参数并验证
        //步骤3：创建一个javasparkcontext对象(ctx)
        //步骤4：使用ctx创建JavaRDD<String>
        //步骤5：JavaRDD<String>创建键值对，其中键是{name},值是{time,value}对
        //步骤6：验证步骤5，打印出来
        //步骤7：按键{name}对JavaRDD<String>元素分组
        //步骤8：验证步骤7，
        //步骤9：对归约器值排序，将得到最终输出
        //步骤10：验证步骤9，
      
        ctx.close();
        System.exit(0);
         
     }
}

- 3.2.1 步骤2:读取输入参数

if (args.length < 2) {
        System.err.println("Usage: SecondarySortUsingGroupByKey <input> <output>");
        System.exit(1);
    }
    String inputPath = args[0];
    System.out.println("inputPath=" + inputPath);
    String outputPath = args[1];
    System.out.println("outputPath=" + outputPath);

- 3.2.2 步骤3:连接到sparkMaster

// STEP-2: Connect to the Spark master by creating JavaSparkContext object
    final JavaSparkContext ctx = SparkUtil.createJavaSparkContext("SecondarySorting");

- 3.2.3 步骤4:创建javaRDD

//  input record format: <name><,><time><,><value>
    JavaRDD<String> lines = ctx.textFile(inputPath, 1);

- 3.2.4 步骤5:从avaRDD中创建键值读，从 <name><,><time><,><value>转换成一个<name，Tuple(time,value)组合>

JavaPairRDD<String, Tuple2<Integer, Integer>> pairs = lines.mapToPair(new PairFunction<String, String, Tuple2<Integer, Integer>>() {
      @Override
      //重写PairFunction中的  Tuple2<K, V> call(T t)方法，此处s 为 T
      public Tuple2<String, Tuple2<Integer, Integer>> call(String s) {
        String[] tokens = s.split(","); // x,2,5
        System.out.println(tokens[0] + "," + tokens[1] + "," + tokens[2]);
        Tuple2<Integer, Integer> timevalue = new Tuple2<Integer, Integer>(Integer.parseInt(tokens[1]), Integer.parseInt(tokens[2]));
        //转换为对应的返回值
        return new Tuple2<String, Tuple2<Integer, Integer>>(tokens[0], timevalue);
      }
    });

- 3.2.5 步骤6:验证步骤五的输出结果

 /**
       *  OUTPUT:
       *  X,2,9
       *  Y,2,5
       *  X,1,3
       *  ……
       */
       
 List<Tuple2<String, Tuple2<Integer, Integer>>> output = pairs.collect();
    for (Tuple2 t : output) {
       Tuple2<Integer, Integer> timevalue = (Tuple2<Integer, Integer>) t._2;
       System.out.println(t._1 + "," + timevalue._1 + "," + timevalue._2);
    }

- 3.2.6 步骤7:通过groupByKey对key进行分组

  // STEP-6: We group JavaPairRDD<> elements by the key ({name}). 
    JavaPairRDD<String, Iterable<Tuple2<Integer, Integer>>> groups = pairs.groupByKey();

- 3.2.7 步骤8:验证步骤七的输入结果

 /**
       *  OUTPUT:
       *  Y
       *  2,5
       *  1,7
       *  3.1    
       *  ……
       */

List<Tuple2<String, Iterable<Tuple2<Integer, Integer>>>> output2 = groups.collect();
    for (Tuple2<String, Iterable<Tuple2<Integer, Integer>>> t : output2) {
       Iterable<Tuple2<Integer, Integer>> list = t._2;
       //输入key
       System.out.println(t._1);
       for (Tuple2<Integer, Integer> t2 : list) {
          //输入value Tuple中的值
          System.out.println(t2._1 + "," + t2._2);
       }
       System.out.println("=====");
    }

- 3.2.8 步骤9:用mapValues对value的第一位进行排序

 //mapValues方法可以对value进行排序，但是不影响key的顺序
    JavaPairRDD<String, Iterable<Tuple2<Integer, Integer>>> sorted = groups.mapValues(new Function<Iterable<Tuple2<Integer, Integer>>,      // input
                                                                                                   Iterable<Tuple2<Integer, Integer>>       // output
                                                                                                  >() {  
      @Override
      public Iterable<Tuple2<Integer, Integer>> call(Iterable<Tuple2<Integer, Integer>> s) {
        List<Tuple2<Integer, Integer>> newList = new ArrayList<Tuple2<Integer, Integer>>(iterableToList(s));
        //SparkTupleComparator中继承了Comparator并重写了它的sort方法
        Collections.sort(newList, SparkTupleComparator.INSTANCE);
        return newList;
      }
    });

- 3.2.9 步骤10:构造结果的打印规则并保持到HDFS

 /**
       *  OUTPUT:
       *  (z,[(1,4),(2,8),(3,7)])
       *  ……
       */

 List<Tuple2<String, Iterable<Tuple2<Integer, Integer>>>> output3 = sorted.collect();
    for (Tuple2<String, Iterable<Tuple2<Integer, Integer>>> t : output3) {
       Iterable<Tuple2<Integer, Integer>> list = t._2;
       System.out.println(t._1);
       for (Tuple2<Integer, Integer> t2 : list) {
          System.out.println(t2._1 + "," + t2._2);
       }
       System.out.println("=====");
    }

    sorted.saveAsTextFile(outputPath);

    System.exit(0);
  }

4.使用scala完成需求

 def main(args: Array[String]): Unit = {
    //
    if (args.length != 3) {
      println("Usage <number-of-partitions> <input-path> <output-path>")
      sys.exit(1)
    }

//    val partitions = args(0).toInt
//    val inputPath = args(1)
//    val outputPath = args(2)

    val partitions = 3
    val inputPath = "C:\\Users\\Administrator\\Desktop\\Book Code\\input.txt"
    val outputPath = " C:\\Users\\Administrator\\Desktop\\Book Code\\output.txt"

    val config = new SparkConf
    config.setAppName("SecondarySort")
    val sc = new SparkContext(config)

    val input = sc.textFile(inputPath)

    //------------------------------------------------
    // each input line/record has the following format:
    // <id><,><time><,><value>
    //-------------------------------------------------
    val valueToKey = input.map(x => {
      val line = x.split(",")
      //返回(<year-day,temperature>,temperature)
      ((line(0) + "-" + line(1), line(2).toInt), line(2).toInt)
    })    
    //隐式转换比较的规则
    implicit def tupleOrderingDesc = new Ordering[Tuple2[String, Int]] {
      override def compare(x: Tuple2[String, Int], y: Tuple2[String, Int]): Int = {
        if (y._1.compare(x._1) == 0) y._2.compare(x._2)
        else y._1.compare(x._1)
      }
    }
    //排序之后的列表
    val sorted = valueToKey.repartitionAndSortWithinPartitions(new CustomPartitioner(partitions))
    //进行结果的转换
    val result = sorted.map {
      case (k, v) => (k._1, v)
    }
    //保持到文件
    result.saveAsTextFile(outputPath)

    // done
    sc.stop()
  }