在mr 框架中,用的最普遍的的一个功能就是比较大小。 map输入时按key排序要比较。reduce的shuff /sort 阶段要比较。
所以我们自定义的key类型都是要实现WritableComparator 接口。这个接口中有三个方法需要我们实现。
先看一个简单的自定义key类型 的代码。我们关注其中有一个
compareTo 方法。
-
package wordcount;
-
-
import java.io.DataInput;
-
import java.io.DataOutput;
-
import java.io.IOException;
-
-
import org.apache.hadoop.io.WritableComparable;
-
-
public class MyKey implements WritableComparable<MyKey> {
-
-
private char c;
-
@Override
-
public void write(DataOutput out) throws IOException {
-
out.writeChar(c);
-
}
-
-
@Override
-
public void readFields(DataInput in) throws IOException {
-
c= in.readChar();
-
}
-
-
@Override
-
public int compareTo(MyKey key) {
-
if(c==key.c)
-
return 0;
-
else if(c> key.c)
-
return 1;
-
else
-
return -1;
-
}
-
-
public char getC() {
-
return c;
-
}
-
-
public void setC(char c) {
-
this.c = c;
-
}
-
@Override
-
public String toString(){
-
return String.valueOf(this.c);
-
}
-
-
}
注意观察我们
public int compareTo(MyKey key) ; 参数是MyKey类型。 就是是说在调用这个比较的时候要做一个反序列化操作成Mykey 对象再比较的过程。 那在大量数据集的情况下,如何减少比较的开销呢。
能不能不反序列化,直接使用序列化成的byte进行比较呢? 答案是肯定的。我们这次介绍的RawComparator 就是用来做这个的。
在真正写我们自己的代码之前 ,先介绍一个小技巧:那就是写任何hadoop MapReduce可编程组件时,先参考系统自带的默认实现代码。
根据这个习惯,我们找一个系统自带的类型看看源码。那就找IntWritable吧。
-
/** A WritableComparable for ints. */
-
@InterfaceAudience.Public
-
@InterfaceStability.Stable
-
public class IntWritable implements WritableComparable<IntWritable> {
-
private int value;
-
-
public IntWritable() {}
-
-
public IntWritable(int value) { set(value); }
-
-
/** Set the value of this IntWritable. */
-
public void set(int value) { this.value = value; }
-
-
/** Return the value of this IntWritable. */
-
public int get() { return value; }
-
-
@Override
-
public void readFields(DataInput in) throws IOException {
-
value = in.readInt();
-
}
-
-
@Override
-
public void write(DataOutput out) throws IOException {
-
out.writeInt(value);
-
}
-
-
/** Returns true iff <code>o</code> is a IntWritable with the same value. */
-
@Override
-
public boolean equals(Object o) {
-
if (!(o instanceof IntWritable))
-
return false;
-
IntWritable other = (IntWritable)o;
-
return this.value == other.value;
-
}
-
-
@Override
-
public int hashCode() {
-
return value;
-
}
-
-
/** Compares two IntWritables. */
-
@Override
-
public int compareTo(IntWritable o) {
-
int thisValue = this.value;
-
int thatValue = o.value;
-
return (thisValue<thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
-
}
-
-
@Override
-
public String toString() {
-
return Integer.toString(value);
-
}
-
-
/** A Comparator optimized for IntWritable. */
-
public static class Comparator extends WritableComparator {
-
public Comparator() {
-
super(IntWritable.class);
-
}
-
-
@Override
-
public int compare(byte[] b1, int s1, int l1,
-
byte[] b2, int s2, int l2) {
-
int thisValue = readInt(b1, s1);
-
int thatValue = readInt(b2, s2);
-
return (thisValue<thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
-
}
-
}
-
-
static { // register this comparator
-
WritableComparator.define(IntWritable.class, new Comparator());
-
}
-
}
从IntWritable 的例子中,我们得到:
1、把WritableComparator 当IntWritable的内部类,写到里面,放到一起比较容易找。
2、如何注册一个自定义的WritableComparator。
此处注意一下,关注static { WritableComparator.define(IntWritable.class,new Comparator())} 的位置。
为什么是写到IntWritable中,写到Comparator类中可以不? 为什么? (不能写到Comparator中)
那按这个方式,我们写一下MyKey的RawComparator
-
package wordcount;
-
-
import java.io.DataInput;
-
import java.io.DataOutput;
-
import java.io.IOException;
-
-
import org.apache.hadoop.io.WritableComparable;
-
import org.apache.hadoop.io.WritableComparator;
-
import org.apache.log4j.Logger;
-
-
public class MyKey implements WritableComparable<MyKey> {
-
-
private char c;
-
@Override
-
public void write(DataOutput out) throws IOException {
-
out.writeChar(c);
-
}
-
-
@Override
-
public void readFields(DataInput in) throws IOException {
-
c= in.readChar();
-
}
-
-
@Override
-
public int compareTo(MyKey key) {
-
if(c==key.c)
-
return 0;
-
else if(c> key.c)
-
return 1;
-
else
-
return -1;
-
}
-
-
public char getC() {
-
return c;
-
}
-
-
public void setC(char c) {
-
this.c = c;
-
}
-
@Override
-
public String toString(){
-
return String.valueOf(this.c);
-
}
-
-
public static class MyKeyRawComparator extends WritableComparator{
-
-
Logger log = Logger.getLogger(MyKeyRawComparator.class);
-
public MyKeyRawComparator(){
-
super(MyKey.class);
-
}
-
-
@Override
-
public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
-
log.info("==============MyKeyRawComparator Used");
-
-
return super.readUnsignedShort(b1,s1) - super.readUnsignedShort(b2, s2);
-
}
-
-
}
-
static{
-
WritableComparator.define(MyKey.class,new MyKeyRawComparator());
-
}
-
-
}
为了验证我们的RawComparator执行了,在compare方法中加了日志输出。同时也可以验证在什么时候调用了两个key的比较。
我们理解,应该在map 与reduce阶段都会使用到key的比较。下面我们就分别看一下map与reduce的日志输出。
打开浏览器输入 http://xxx.xxx.xxx.xxx:19888/jobhistory 点击对应的job.
mapper的日志

reducer的日志

通过日志,验证了我们的预期。
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/30066956/viewspace-2112283/
,如需转载,请注明出处,否则将追究法律责任。