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 http://www.linuxidc.com/Linux/2013-05/84948.htm

Hadoop权威指南（中文第2版）PDF http://www.linuxidc.com/Linux/2012-07/65972.htm

采用MapReduce与Hadoop进行大数据分析 http://www.linuxidc.com/Linux/2013-07/87312.htm