这似乎是一个愚蠢的问题,但我在 hadoop 的 mapreduce 代码中没有看到我的类型中的问题
正如问题中所述,问题是它需要 IntWritable,但我在减速器的collector.collect 中向它传递了一个 Text 对象。
我的作业配置具有以下映射器输出类:
conf.setMapOutputKeyClass(IntWritable.class);
conf.setMapOutputValueClass(IntWritable.class);
以及以下减速器输出类:
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);
我的映射类具有以下定义:
public static class Reduce extends MapReduceBase implements Reducer<IntWritable, IntWritable, Text, IntWritable>
具有所需的功能:
public void reduce(IntWritable key, Iterator<IntWritable> values, OutputCollector<Text,IntWritable> output, Reporter reporter)
然后当我打电话时它失败了:
output.collect(new Text(),new IntWritable());
我对映射归约相当陌生,但所有类型似乎都匹配,它编译但随后在该行失败,表示它期望 IntWritable 作为归约类的键。如果重要的话我正在使用 0.21 版本的 Hadoop
这是我的地图类:
public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, IntWritable, IntWritable> {
private IntWritable node = new IntWritable();
private IntWritable edge = new IntWritable();
public void map(LongWritable key, Text value, OutputCollector<IntWritable, IntWritable> output, Reporter reporter) throws IOException {
String line = value.toString();
StringTokenizer tokenizer = new StringTokenizer(line);
while (tokenizer.hasMoreTokens()) {
node.set(Integer.parseInt(tokenizer.nextToken()));
edge.set(Integer.parseInt(tokenizer.nextToken()));
if(node.get() < edge.get())
output.collect(node, edge);
}
}
}
和我的减课:
public static class Reduce extends MapReduceBase implements Reducer<IntWritable, IntWritable, Text, IntWritable> {
IntWritable $ = new IntWritable(Integer.MAX_VALUE);
Text keyText = new Text();
public void reduce(IntWritable key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
ArrayList<IntWritable> valueList = new ArrayList<IntWritable>();
//outputs original edge pair as key and $ for value
while (values.hasNext()) {
IntWritable value = values.next();
valueList.add(value);
keyText.set(key.get() + ", " + value.get());
output.collect(keyText, $);
}
//outputs all the 2 length pairs
for(int i = 0; i < valueList.size(); i++)
for(int j = i+1; i < valueList.size(); j++)
output.collect(new Text(valueList.get(i).get() + ", " + valueList.get(j).get()), key);
}
}
和我的工作配置:
JobConf conf = new JobConf(Triangles.class);
conf.setJobName("mapred1");
conf.setMapOutputKeyClass(IntWritable.class);
conf.setMapOutputValueClass(IntWritable.class);
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(IntWritable.class);
conf.setMapperClass(Map.class);
conf.setCombinerClass(Reduce.class);
conf.setReducerClass(Reduce.class);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path("mapred1"));
JobClient.runJob(conf);