MapReduce的自制Writable分组输出及组内排序

MapReduce的自制Writable分组输出及组内排序

问题描述：

输入文件格式如下：
name1    2
name3    4
name1    6
name1    1
name3    3
name1    0
要求输出的文件格式如下：
name1    0，1，2，6
name3    3，4
要求是按照第一列分组，name1与name3也是按照顺序排列的，组内升序排序。

思路：
常规的输出，无法排序key所对应的多个值的顺序。为了排序组内中的值，需要将key与value放在同一个组。Job中有两个方法setGroupingComparatorClass和setSortComparatorClass，可以利用这两个方法来实现组内排序。但是这些排序都是基于key的，则就要将key和value定义成组合键。
但是必须要保证第一列相同的全部都放在同一个分区中，则就需要自定义分区，分区的时候只考虑第一列的值。由于partitioner仅仅能保证每一个reducer接受同一个name的所有记录，但是reducer仍然是通过键进行分组的分区，也就说该分区中还是按照键来分成不同的组，还需要分组只参考name值
先按照name分组，再在name中内部进行排序。

解决方法：
运用自定义组合键的策略，将name和1定义为一个组合键。在分区的时候只参考name的值，即继承partitioner。
 由于要按照name分组，则就需要定义分组策略，然后设置setGroupingComparatorClass。
setGroupingComparatorClass主要定义哪些key可以放置在一组，分组的时候会对组合键进行比较，由于这里只需要考虑组合键中的一个值，则定义实现一个WritableComparator，设置比较策略。
对于组内的排序，可以利用setSortComparatorClass来实现，
这个方法主要用于定义key如何进行排序在它们传递给reducer之前，
这里就可以来进行组内排序。
具体代码：
    Hadoop版本号：hadoop1.1.2
自定义组合键

package whut;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
//自定义组合键策略
//java基本类型数据
public class TextInt implements WritableComparable{
    //直接利用java的基本数据类型
    private String firstKey;
    private int secondKey;
    //必须要有一个默认的构造函数
    public String getFirstKey() {
        return firstKey;
    }
    public void setFirstKey(String firstKey) {
        this.firstKey = firstKey;
    }
    public int getSecondKey() {
        return secondKey;
    }
    public void setSecondKey(int secondKey) {
        this.secondKey = secondKey;
    }
                                                                                                                                                                         
    @Override
    public void write(DataOutput out) throws IOException {
        // TODO Auto-generated method stub
        out.writeUTF(firstKey);
        out.writeInt(secondKey);
    }
    @Override
    public void readFields(DataInput in) throws IOException {
        // TODO Auto-generated method stub
        firstKey=in.readUTF();
        secondKey=in.readInt();
    }
    //map的键的比较就是根据这个方法来进行的
    @Override
    public int compareTo(Object o) {
        // TODO Auto-generated method stub
        TextInt ti=(TextInt)o;
        //利用这个来控制升序或降序
        //this本对象写在前面代表是升序
        //this本对象写在后面代表是降序
        return this.getFirstKey().compareTo(ti.getFirstKey());
    }
}

分组策略

package whut;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
//主要就是对于分组进行排序，分组只按照组建键中的一个值进行分组
public class TextComparator extends WritableComparator {
    //必须要调用父类的构造器
    protected TextComparator() {
        super(TextInt.class,true);//注册comparator
    }
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        // TODO Auto-generated method stub
        TextInt ti1=(TextInt)a;
        TextInt ti2=(TextInt)b;
        return ti1.getFirstKey().compareTo(ti2.getFirstKey());
    }
}

Hadoop权威指南(中文版-带目录索引)PDF
Hadoop权威指南（中文第2版）PDF
采用MapReduce与Hadoop进行大数据分析