/** Combines values for a given key.
* @param key the key is expected to be a Text object, whose prefix indicates
* the type of aggregation to aggregate the values.
* @param values the values to combine
* @param context to collect combined values
*/
public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
String keyStr = key.toString();
int pos = keyStr.indexOf(ValueAggregatorDescriptor.TYPE_SEPARATOR);
String type = keyStr.substring(0, pos);
long uniqCount = context.getConfiguration().
getLong(UniqValueCount.MAX_NUM_UNIQUE_VALUES, Long.MAX_VALUE);
ValueAggregator aggregator = ValueAggregatorBaseDescriptor
.generateValueAggregator(type, uniqCount);
for (Text val : values) {
aggregator.addNextValue(val);
}
Iterator<?> outputs = aggregator.getCombinerOutput().iterator();
while (outputs.hasNext()) {
Object v = outputs.next();
if (v instanceof Text) {
context.write(key, (Text)v);
} else {
context.write(key, new Text(v.toString()));
}
}
}
ValueAggregatorCombiner.java 文件源码
java
阅读 22
收藏 0
点赞 0
评论 0
项目:hadoop
作者:
评论列表
文章目录