目标：弄明白，我们配置的GroupComparator是如何对进入reduce函数中的key Iterable<value> 进行影响。
如下是一个配置了GroupComparator 的reduce 函数。具体影响是我们可以在自定义的GroupComparator 中确定哪儿些value组成一组，进入一个reduce函数

点击(此处)折叠或打开

public static class DividendGrowthReducer extends Reducer<Stock, DoubleWritable, NullWritable, DividendChange> {
private NullWritable outputKey = NullWritable.get();
private DividendChange outputValue = new DividendChange();
@Override
protected void reduce(Stock key, Iterable<DoubleWritable> values, Context context)
throws IOException, InterruptedException {
double previousDividend = 0.0;
for(DoubleWritable dividend : values) {
double currentDividend = dividend.get();
double growth = currentDividend - previousDividend;
if(Math.abs(growth) > 0.000001) {
outputValue.setSymbol(key.getSymbol());
outputValue.setDate(key.getDate());
outputValue.setChange(growth);
context.write(outputKey, outputValue);
previousDividend = currentDividend;
}
}
}
}

着先我们找到向上找，是谁调用了我们写的这个reduce函数。 Reducer类的run 方法。通过如下代码，可以看到是在run方法中，对于每个key，调用一次reduce函数。
此处传入reduce函数的都是对象引用。

点击(此处)折叠或打开

/**
* Advanced application writers can use the
* {@link #run(org.apache.hadoop.mapreduce.Reducer.Context)} method to
* control how the reduce task works.
*/
public void run(Context context) throws IOException, InterruptedException {
.....
while (context.nextKey()) {
reduce(context.getCurrentKey(), context.getValues(), context);
.....
}
.....
}
}

结合我们写的reduce函数，key是在遍历value的时候会对应变化。
那我们继续跟踪context.getValues 得到的迭代器的next方法。context 此处是ReduceContext.java （接口）. 对应的实现类为ReduceContextImpl.java

点击(此处)折叠或打开

protected class ValueIterable implements Iterable<VALUEIN> {
private ValueIterator iterator = new ValueIterator();
@Override
public Iterator<VALUEIN> iterator() {
return iterator;
}
}
/**
* Iterate through the values for the current key, reusing the same value
* object, which is stored in the context.
* @return the series of values associated with the current key. All of the
* objects returned directly and indirectly from this method are reused.
*/
public
Iterable<VALUEIN> getValues() throws IOException, InterruptedException {
return iterable;
}

直接返回了一个iterable。继续跟踪ValueIterable 类型的iterable。那明白了，在reduce 函数中进行Iterable的遍历，其实调用的是ValueIterable的next方法。下面看一下next的实现。

点击(此处)折叠或打开

@Override
public VALUEIN next() {
………………
nextKeyValue();
return value;
………………
}

再继续跟踪nextKeyValue()方法。终于找了一个comparator。这个就是我们配置的GroupingComparator.

点击(此处)折叠或打开

@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
……………………………………
if (hasMore) {
nextKey = input.getKey();
nextKeyIsSame = comparator.compare(currentRawKey.getBytes(), 0,
currentRawKey.getLength(),
nextKey.getData(),
nextKey.getPosition(),
nextKey.getLength() - nextKey.getPosition()
) == 0;
} else {
nextKeyIsSame = false;
}
inputValueCounter.increment(1);
return true;
}

为了证明这个就是我们配置的GroupingComparator。跟踪ReduceContextImpl的构造调用者。 ReduceTask的run方法。

点击(此处)折叠或打开

@Override
@SuppressWarnings("unchecked")
public void run(JobConf job, final TaskUmbilicalProtocol umbilical){
………………………………
RawComparator comparator = job.getOutputValueGroupingComparator();
runNewReducer(job, umbilical, reporter, rIter, comparator,
keyClass, valueClass);
}

下面把runNewReducer 的代码也贴出来。

点击(此处)折叠或打开

void runNewReducer(JobConf job,
final TaskUmbilicalProtocol umbilical,
final TaskReporter reporter,
RawKeyValueIterator rIter,
RawComparator<INKEY> comparator,
Class<INKEY> keyClass,
Class<INVALUE> valueClass
) {
org.apache.hadoop.mapreduce.Reducer.Context
reducerContext = createReduceContext(reducer, job, getTaskID(),
rIter, reduceInputKeyCounter,
reduceInputValueCounter,
trackedRW,
committer,
reporter, comparator, keyClass,
valueClass);

好吧，关于自定义GroupingComparator如何起做用的代码分析，就到此吧。

ITPub博客

Hadoop的GroupComparator是如何起做用的（源码分析）

最新文章