写在前面
本篇博客背景
? 在使用Mapreduce程序时,不难看出Map阶段的输出键值对其实是默认按照字典排序的,而我们在使用该程序的时候也会因为需要满足某种需求自定义排序规则。当前我们有两种实现自定义排序的方式:
两种实现自定义排序的方式
-
直接让参与排序的对象实现WritableComparable接口,并在接口中实现compareTo方法,当运行的时候hadoop会自动帮助我们生成WritableComparator对象。 具体可见下例: phoneBean.java package com.yxp.complication.exercise.sorttest;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class PhoneBean implements WritableComparable<PhoneBean> {
public int upFlow;
public int downFlow;
public int sumFlow;
public int getUpFlow() {
return upFlow;
}
public void setUpFlow(int upFlow) {
this.upFlow = upFlow;
}
public int getDownFlow() {
return downFlow;
}
public void setDownFlow(int downFlow) {
this.downFlow = downFlow;
}
public int getSumFlow() {
return sumFlow;
}
public void setSumFlow() {
this.sumFlow=upFlow+downFlow;
}
@Override
public void write(DataOutput out) throws IOException {
out.writeInt(upFlow);
out.writeInt(downFlow);
out.writeInt(sumFlow);
}
@Override
public void readFields(DataInput in) throws IOException {
this.upFlow = in.readInt();
this.downFlow=in.readInt();
this.sumFlow=in.readInt();
}
@Override
public String toString() {
return upFlow + "\t" + downFlow + "\t" + sumFlow ;
}
@Override
public int compareTo(PhoneBean o) {
return -(this.sumFlow-o.sumFlow);
}
}
-
自定义一个比较器对象,继承WritableComparator类并重写compare方法,在该方法中定义比较规则,注意在自定义的比较器对象中通过调用父类的super方法将自定义的比较器对象和要参与比较的对象关联。最后在Driver类中设置该比较器类为作业的比较器类。
具体可见下例:
PhoneBeanComparator.java
package com.yxp.complication.exercise.sorttest;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
public class PhoneBeanComparator extends WritableComparator {
public PhoneBeanComparator() {
super(PhoneBean.class,true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
PhoneBean abean = (PhoneBean) a;
abean.setSumFlow();
PhoneBean bbean = (PhoneBean) b;
bbean.setSumFlow();
return -(abean.sumFlow - bbean.sumFlow);
}
}
我的疑问
疑问便是第一种实现方式是如何得到排序器对象从而排序的,因为它根本没有实例化啊!虽然结论是Hadoop帮他构造的,但是只知道这个未免有些粗浅了,因此我们要找到源头!
探索底层规则,记得找源码,看源码,理解源码。
思路展开
我们都知道使用job对象可以设置提交作业的各种参数,如:Mapper类、Reducer类、Map端的输出键值对类型,程序的输出键值对类型……由此自然而然的联想到程序的比较器是不是也可以在这里设置呢?
寻找过程
getSortComparator()–>
我尝试着使用job.comparator然后聪明的IEDA就为我找到了下面的这个方法。
ctrl+鼠标左键进入该方法,方法体如下:
public RawComparator<?> getSortComparator() {
return conf.getOutputKeyComparator();
}
第二行描述足够明显,于是点进7行中的getOutputKeyComparator()方法——方法体如下:
public RawComparator getOutputKeyComparator() {
Class<? extends RawComparator> theClass = getClass(
JobContext.KEY_COMPARATOR, null, RawComparator.class);
if (theClass != null)
return ReflectionUtils.newInstance(theClass, this);
return WritableComparator.get(getMapOutputKeyClass().asSubclass(WritableComparable.class), this);
}
在这里逻辑已经相对具体了,我们把聚焦在getClass()的三个参数上。
第一个:JobContext.KEY_COMPARATOR,关于这个变量的定义是这样的:
public static final String KEY_COMPARATOR = "mapreduce.job.output.key.comparator.class";
第二个:null,作为第一个变量的默认值。
第三个:RawComparator.class,直接对对象的字节表示进行操作的Comparator 。
它出现在这里是什么意思呢?或者说:
getClass()的三个参数代表什么意义呢?还是要走进去看一看。
* @param name the class name. (类名)
* @param defaultValue default value.(默认值)
* @param xface the interface implemented by the named class.(由类名实现的接口)
* @return property value as a <code>Class</code>,
* or <code>defaultValue</code>.
*/
public <U> Class<? extends U> getClass(String name,
Class<? extends U> defaultValue,
Class<U> xface)
所以我们可以理解了,getClass()的功能是获取比较器的类,如果用户设置了就使用设置的类,没有设置的话值为null。
那么下面的代码看起来也相对好理解一些了。
1-2行:如果不为null(设置了比较器的类),就使用反射的工具类,将这个类实例化。
3行:返回WritableComparator的get()方法的返回值。这是建立在上面的判断不成立的情况,也就是说如果我们没有设置比较器类就会调用该语句,我仿佛接近真相了。
该方法的两个参数:
-
getMapOutputKeyClass().asSubclass(WritableComparable.class)作用是 获取Map的输出key的类,将其作为WritableComparable类的子类 -
this表示是该作业的conf对象。
if (theClass != null)
return ReflectionUtils.newInstance(theClass, this);
return WritableComparator.get(getMapOutputKeyClass().asSubclass(WritableComparable.class), this);
走进get方法体——>
public static WritableComparator get(
Class<? extends WritableComparable> c, Configuration conf) {
WritableComparator comparator = comparators.get(c);
if (comparator == null) {
forceInit(c);
comparator = comparators.get(c);
if (comparator == null) {
comparator = new WritableComparator(c, conf, true);
}
}
ReflectionUtils.setConf(comparator, conf);
return comparator;
}
总结
Mapreduce程序在对Key排序时,
如果已经给Job对象设置好了比较器的类,会为它创建实例并按照其中的排序规则对Key进行排序,此为第二种自定义排序实现方式的逻辑。
否则,会从Hadoop自带的数据类型对应的比较器集合中寻找当前需要比较对象的比较器,
? 如果没有找到,为了防止极端情况发生导致原本该存在的比较器被回收,会进行一次强制初始化,然后再从集合中寻找它。
? 这一次如果再找不到,会判断该对象不是Hadoop自带的数据类型,如果需要比较的类实现了WritableComparable接口,则为它创造一个WritableComparator对象实例,使第二种实现方式有了等同第一种实现方式的效果。
|