全部
我有简单的 map/reduce 实现。调用 Mapper 并完成其工作,但从不调用 reducer。
这是映射器:
static public class InteractionMap extends Mapper<LongWritable, Text, Text, InteractionWritable> {
@Override
protected void map(LongWritable offset, Text text, Context context) throws IOException, InterruptedException {
System.out.println("mapper");
String[] tokens = text.toString().split(",");
for (int idx = 0; idx < tokens.length; idx++) {
String sourceUser = tokens[1];
String targetUser = tokens[2];
int points = Integer.parseInt(tokens[4]);
context.write(new Text(sourceUser), new InteractionWritable(targetUser, points));
}
}
}
}
这是我的减速器:
static public class InteractionReduce extends Reducer<Text, InteractionWritable, Text, Text> {
@Override
protected void reduce(Text token, Iterable<InteractionWritable> counts, Context context) throws IOException, InterruptedException {
System.out.println("REDUCER");
Iterator<InteractionWritable> i = counts.iterator();
while (i.hasNext()) {
InteractionWritable interaction = i.next();
context.write(token, new Text(token.toString() + " " + interaction.getTargetUser().toString() + " " + interaction.getPoints().get()));
}
}
}
而且,这是配置部分:
@Override
public int run(String[] args) throws Exception {
Configuration configuration = getConf();
Job job = new Job(configuration, "Interaction Count");
job.setJarByClass(InteractionMapReduce.class);
job.setMapperClass(InteractionMap.class);
job.setCombinerClass(InteractionReduce.class);
job.setReducerClass(InteractionReduce.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
return job.waitForCompletion(true) ? 0 : -1;
}
有谁知道为什么没有调用减速器?