基于MapReduce思想，编写SecondarySort程序。
要能理解MapReduce编程思想，会编写MapReduce版本二次排序程序，然后将其执行并分析执行过程。
MR默认会对键进行排序，然而有的时候我们也有对值进行排序的需求。满足这种需求一是可以在reduce阶段排序收集过来的values，但是，如果有数量巨大的values可能就会导致内存溢出等问题，这就是二次排序应用的场景——将对值的排序也安排到MR计算过程之中，而不是单独来做。
二次排序就是首先按照第一字段排序，然后再对第一字段相同的行按照第二字段排序，注意不能破坏第一次排序的结果。
程序主要难点在于排序和聚合。
对于排序我们需要定义一个IntPair类用于数据的存储，并在IntPair类内部自定义Comparator类以实现第一字段和第二字段的比较。
对于聚合我们需要定义一个FirstPartitioner类，在FirstPartitioner类内部指定聚合规则为第一字段。
此外，我们还需要开启MapReduce框架自定义Partitioner 功能和GroupingComparator功能。
IntPair 类：
package mr;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.WritableComparable;
public class IntPair implements WritableComparable {
private IntWritable first;
private IntWritable second;
public void set(IntWritable first, IntWritable second) {
this.first = first;
this.second = second;
}
//注意：需要添加无参的构造方法，否则反射时会报错。
public IntPair() {
set(new IntWritable(), new IntWritable());
}
public IntPair(int first, int second) {
set(new IntWritable(first), new IntWritable(second));
}
public IntPair(IntWritable first, IntWritable second) {
set(first, second);
}
public IntWritable getFirst() {
return first;
}
public void setFirst(IntWritable first) {
this.first = first;
}
public IntWritable getSecond() {
return second;
}
public void setSecond(IntWritable second) {
this.second = second;
}
public void write(DataOutput out) throws IOException {
first.write(out);
second.write(out);
}
public void readFields(DataInput in) throws IOException {
first.readFields(in);
second.readFields(in);
}
public int hashCode() {
return first.hashCode() * 163 + second.hashCode();
}
public boolean equals(Object o) {
if (o instanceof IntPair) {
IntPair tp = (IntPair) o;
return first.equals(tp.first) && second.equals(tp.second);
}
return false;
}
public String toString() {
return first + "\t" + second;
}
public int compareTo(IntPair tp) {
int cmp = first.compareTo(tp.first);
if (cmp != 0) {
return cmp;
}
return second.compareTo(tp.second);
}
}完整代码：
package mr;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class SecondarySort {
static class TheMapper extends Mapper<LongWritable, Text, IntPair, NullWritable> {
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] fields = value.toString().split("\t");
int field1 = Integer.parseInt(fields[0]);
int field2 = Integer.parseInt(fields[1]);
context.write(new IntPair(field1,field2), NullWritable.get());
}
}
static class TheReducer extends Reducer<IntPair, NullWritable,IntPair, NullWritable> {
//private static final Text SEPARATOR = new Text("------------------------------------------------");
@Override
protected void reduce(IntPair key, Iterable<NullWritable> values, Context context)
throws IOException, InterruptedException {
context.write(key, NullWritable.get());
}
}
public static class FirstPartitioner extends Partitioner<IntPair, NullWritable> {
public int getPartition(IntPair key, NullWritable value,
int numPartitions) {
return Math.abs(key.getFirst().get()) % numPartitions;
}
}
//如果不添加这个类，默认第一列和第二列都是升序排序的。
//这个类的作用是使第一列升序排序，第二列降序排序
public static class KeyComparator extends WritableComparator {
//无参构造器必须加上，否则报错。
protected KeyComparator() {
super(IntPair.class, true);
}
public int compare(WritableComparable a, WritableComparable b) {
IntPair ip1 = (IntPair) a;
IntPair ip2 = (IntPair) b;
//第一列按升序排序
int cmp = ip1.getFirst().compareTo(ip2.getFirst());
if (cmp != 0) {
return cmp;
}
//在第一列相等的情况下，第二列按倒序排序
return -ip1.getSecond().compareTo(ip2.getSecond());
}
}
//入口程序
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
job.setJarByClass(SecondarySort.class);
//设置Mapper的相关属性
job.setMapperClass(TheMapper.class);
//当Mapper中的输出的key和value的类型和Reduce输出
//的key和value的类型相同时，以下两句可以省略。
//job.setMapOutputKeyClass(IntPair.class);
//job.setMapOutputValueClass(NullWritable.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
//设置分区的相关属性
job.setPartitionerClass(FirstPartitioner.class);
//在map中对key进行排序
job.setSortComparatorClass(KeyComparator.class);
//job.setGroupingComparatorClass(GroupComparator.class);
//设置Reducer的相关属性
job.setReducerClass(TheReducer.class);
job.setOutputKeyClass(IntPair.class);
job.setOutputValueClass(NullWritable.class);
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//设置Reducer数量
int reduceNum = 1;
if(args.length >= 3 && args[2] != null){
reduceNum = Integer.parseInt(args[2]);
}
job.setNumReduceTasks(reduceNum);
job.waitForCompletion(true);
}
}使用Eclipse开发工具将该代码打包，选择主类为mr.Secondary。如果没有指定主类，那么在执行时就要指定须执行的类。假定打包后的文件名为Secondary.jar，主类SecondarySort位于包mr下，则可使用如下命令向Hadoop集群提交本应用。
[root@master hadoop]# bin/hadoop jar SecondarySort.jar mr.Secondary /user/mapreduce/secsort/in/secsortdata.txt /user/mapreduce/secsort/out 1其中“hadoop”为命令，“jar”为命令参数，后面紧跟打的包，/user/mapreduce/secsort/in/secsortdata.txt”为输入文件在HDFS中的位置，如果HDFS中没有这个文件，则自己自行上传。“/user/mapreduce/secsort/out/”为输出文件在HDFS中的位置，“1”为Reduce个数。
输入数据如下：secsortdata.txt ('\t'分割)（数据放在/root/data/6目录下）：
7 444
3 9999
7 333
4 22
3 7777
7 555
3 6666
6 0
3 8888
4 11在master上执行对hdfs上的文件/user/mapreduce/secsort/out/part-r-00000内容查看的操作
[root@master hadoop]# bin/hadoop fs -cat /user/mapreduce/secsort/out/p*如图6-1所示：
图6-1
步骤1：搭建Hadoop集群
步骤2：上传数据文件至HDFS
步骤3：编写IntPair程序
步骤4：编写SecondarySort程序
步骤5：打包程序
步骤6：运行程序
步骤7：查看运行结果
