java - 错误的 key 类 : Text is not IntWritable

Question

这看起来像是一个愚蠢的问题，但我在我的 hadoop mapreduce 代码中没有看到我的类型中的问题

如问题中所述，问题在于它期望 IntWritable 但我在 reducer 的 collector.collect 中将一个 Text 对象传递给它。

我的作业配置具有以下映射器输出类:

conf.setMapOutputKeyClass(IntWritable.class);
conf.setMapOutputValueClass(IntWritable.class);

以及以下 reducer 输出类:

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());

我是 map reduce 的新手，但所有类型似乎都匹配，它编译但随后在该行失败，说它期望 IntWritable 作为 reduce 类的键。如果这很重要，我使用的是 0.21 版本的 Hadoop

这是我的 map 类:

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);
        }
    }
}

和我的 reduce 类:

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);

Answer 1

您的问题是您将 Reduce 类设置为组合器

conf.setCombinerClass(Reduce.class);

组合器在映射阶段运行，它们需要发出相同的键/值类型(在您的情况下为 IntWriteable、IntWritable)删除这一行，你应该没问题