MapReduce排序

• 在map和reduce阶段进行排序时，比较的是k2。v2是不参与排序比较的。如果要想让v2也进行排序，需要把k2和v2组装成新的类，作为k2，才能参与比较。

例子:

    

二次排序：在第一列有序得到前提下第二列進行排序。

思路：先找<k3,v3>在找<k2,v2>之後的mapreduce就容易寫了

方法1：让输出的第一列作为k3，第二列作为v3    关键：输出的v3需要参与排序，此种方式无法实现二次排序

方法2：让1,2列只作为k3，而v3为空。(

方法3：有可能让k3为空，v3为第二列吗？ 答案是不能的，假设k3为空，一般情况下k2也为空，则v2中存放的数据进入后每一组都会放入一个value中，目前没有遇到)

      因此，只能选择方法二进行二次排序。

根据前面知识，关键思路：排序和分组是按照k2进行排序和分组的情形需铭记。

第一部分：分部代码

自定义排序：    

private static class TwoInt implements WritableComparable<TwoInt>{

    public int t1;
    public int t2;
    public void write(DataOutput out) throws IOException {
        out.writeInt(t1);        
        out.writeInt(t2);        
    }

    public void set(int t1, int t2) {
        this.t1=t1;
        this.t2=t2;
    }

    public void readFields(DataInput in) throws IOException {
        this.t1=in.readInt();
        this.t2=in.readInt();
    }

    
    public int compareTo(TwoInt o) {
        if (this.t1 ==o.t1) { //當第一列相等的時候，第二列升序排列
            return this.t2 -o.t2;
        }
        return this.t1-o.t1;//當第一列不相等的時候，按第一列升序排列
    }
    
}

自定义Mapper类

private static class MyMapper extends Mapper<LongWritable, Text, TwoInt, NullWritable>{
    TwoInt K2 = new TwoInt();
    @Override
    protected void map(LongWritable key, Text value,
            Mapper<LongWritable, Text, TwoInt, NullWritable>.Context context)
            throws IOException, InterruptedException {
           String[] splited = value.toString().split("	");
           K2.set(Integer.parseInt(splited[0]),Integer.parseInt(splited[1]));
           context.write(K2, NullWritable.get());
    }
}

自定义Reduce类

//按照k2進行排序，分組，此數據分爲6組，在調用Reduce
private static class MyReducer extends Reducer<TwoInt, NullWritable, TwoInt, NullWritable>{
    @Override
    protected void reduce(TwoInt k2, Iterable<NullWritable> v2s,
            Reducer<TwoInt, NullWritable, TwoInt, NullWritable>.Context context)
            throws IOException, InterruptedException {
        context.write(k2, NullWritable.get());
    }
}

捆绑Map和Reduce在一起

public static void main(String[] args) throws Exception {
    Job job = Job.getInstance(new Configuration(), SecondarySortTest.class.getSimpleName());
    job.setJarByClass(SecondarySortTest.class);
    //1.自定义输入路径
    FileInputFormat.setInputPaths(job, new Path(args[0]));
    //2.自定义mapper
    //job.setInputFormatClass(TextInputFormat.class);
    job.setMapperClass(MyMapper.class);
    //job.setMapOutputKeyClass(Text.class);
    //job.setMapOutputValueClass(TrafficWritable.class);
    
    //3.自定义reduce
    job.setReducerClass(MyReducer.class);
    job.setOutputKeyClass(TwoInt.class);
    job.setOutputValueClass(NullWritable.class);
    //4.自定义输出路径
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    //job.setOutputFormatClass(TextOutputFormat.class);//对输出的数据格式化并写入磁盘
    
    job.waitForCompletion(true);
}

由此，可以完成二次排序的完整代码如下：

package Mapreduce;

import java.io.DataInput;
import java.io.DataOutput;
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.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SecondarySortTest {
public static void main(String[] args) throws Exception {
    Job job = Job.getInstance(new Configuration(), SecondarySortTest.class.getSimpleName());
    job.setJarByClass(SecondarySortTest.class);
    //1.自定义输入路径
    FileInputFormat.setInputPaths(job, new Path(args[0]));
    //2.自定义mapper
    //job.setInputFormatClass(TextInputFormat.class);
    job.setMapperClass(MyMapper.class);
    //job.setMapOutputKeyClass(Text.class);
    //job.setMapOutputValueClass(TrafficWritable.class);
    
    //3.自定义reduce
    job.setReducerClass(MyReducer.class);
    job.setOutputKeyClass(TwoInt.class);
    job.setOutputValueClass(NullWritable.class);
    //4.自定义输出路径
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    //job.setOutputFormatClass(TextOutputFormat.class);//对输出的数据格式化并写入磁盘
    
    job.waitForCompletion(true);
}
private static class MyMapper extends Mapper<LongWritable, Text, TwoInt, NullWritable>{
    TwoInt K2 = new TwoInt();
    @Override
    protected void map(LongWritable key, Text value,
            Mapper<LongWritable, Text, TwoInt, NullWritable>.Context context)
            throws IOException, InterruptedException {
           String[] splited = value.toString().split("	");
           K2.set(Integer.parseInt(splited[0]),Integer.parseInt(splited[1]));
           context.write(K2, NullWritable.get());
    }
}
//按照k2進行排序，分組，此數據分爲6組，在調用Reduce
private static class MyReducer extends Reducer<TwoInt, NullWritable, TwoInt, NullWritable>{
    @Override
    protected void reduce(TwoInt k2, Iterable<NullWritable> v2s,
            Reducer<TwoInt, NullWritable, TwoInt, NullWritable>.Context context)
            throws IOException, InterruptedException {
        context.write(k2, NullWritable.get());
    }
}

private static class TwoInt implements WritableComparable<TwoInt>{
    public int t1;
    public int t2;
    public void write(DataOutput out) throws IOException {
        out.writeInt(t1);        
        out.writeInt(t2);        
    }
    public void set(int t1, int t2) {
        this.t1=t1;
        this.t2=t2;
    }
    public void readFields(DataInput in) throws IOException {
        this.t1=in.readInt();
        this.t2=in.readInt();
    }
    public int compareTo(TwoInt o) {
        if (this.t1 ==o.t1) { //當第一列相等的時候，第二列升序排列
            return this.t2 -o.t2;
        }
        return this.t1-o.t1;//當第一列不相等的時候，按第一列升序排列
    }
    @Override
    public String toString() {
        return t1+"	"+t2;
    }
}
}

第二部分：测试代码

（1）准备环境，准备测试数据

[root@neusoft-master filecontent]# vi twoint
3         3
3         2
3         1
2         2
2         1
1         1

 （2）创建文件夹，并将文件上传到HDFS中

[root@neusoft-master filecontent]# hadoop dfs -mkdir /neusoft/

[root@neusoft-master filecontent]# hadoop dfs -put twoint /neusoft/

 

(3)执行jar包，查看中间过程

 [root@neusoft-master filecontent]# hadoop jar SecondarySortTest.jar /neusoft/twoint  /out8

 

（4）查看结果

[root@neusoft-master filecontent]# hadoop dfs -ls /out8

[root@neusoft-master filecontent]# hadoop dfs -text /out8/part-r-00000

 

结果正确。

• 如果输出有错误的话，或者输出不是数字(有时候是对象)，需要查看是否重写了tostring()方法

 注意：如果需求变更为第一列的升序和第二列的降序，只需更改第3行

    public int compareTo(TwoInt o) {
        if (this.t1 ==o.t1) { //當第一列相等的時候，第二列降序排列
            return o.t2-this.t2;
 } 5 return this.t1-o.t1;//當第一列不相等的時候，按第一列升序排列 6 }

 总结：value不能参与排序，如果想参加排序需要放在key中，作为一个新的key进行排序。