首页 文章

Reducer多次接收相同的值而不是预期的输入

提问于
浏览
0

在我的本地hadoop环境中编写map-reduce作业时,我遇到了Reducer没有收到我期望的值的问题 . 我将问题抽象为以下内容:

我用10行创建一个任意输入文件,让map方法执行10次 . 在映射器中,我创建一个调用计数,并将此计数值作为值写入输出,如果值为偶数,则为0作为键,如果值为奇数,则为1作为键,即以下(键,值)对:

(1,1),(0,2),(1,3),(0,4),(1,5)等

我希望收到两次对Reducer的调用

  • 0> [2,4,6,8,10]

  • 1> [1,3,5,7,9]

但我接到两个电话

  • 0> [2,2,2,2,2]

  • 1> [1,1,1,1,1]

代替 . 我似乎收到了映射器中写入的第一个带有键的多重性的值(如果我反转计数器,我会得到值10和9而不是2和1) . 根据我的理解,这不是预期的行为(?),但我无法弄清楚我做错了什么 .

我使用以下Mapper和reducer:

public class TestMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
    int count = 0;

    @Override
    protected void map(LongWritable keyUnused, Text valueUnused, Context context) throws IOException, InterruptedException {
        count += 1;
        context.write(new IntWritable(count % 2), new IntWritable(count));

        System.err.println((count % 2) + "|" + count);
    }
}

public class TestReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
    @Override
    protected void reduce(IntWritable key, Iterable<IntWritable> valueItr, Context context) throws IOException, InterruptedException {
        List<IntWritable> values = Lists.newArrayList(valueItr);

        System.err.println(key + "|" + values);
    }
}

我使用本地测试运行器运行hadoop作业,如“Hadoop:The Definitive Guide”(O'Reilly)一书中所述:

public class TestDriver extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.printf("Usage: %s [generic options] <input> <output>\n",
                    getClass().getSimpleName());
            ToolRunner.printGenericCommandUsage(System.err);
            return -1;
        }

        Job jobConf = Job.getInstance(getConf());
        jobConf.setJarByClass(getClass());
        jobConf.setJobName("TestJob");  

        jobConf.setMapperClass(TestMapper.class);
        jobConf.setReducerClass(TestReducer.class);

        FileInputFormat.addInputPath(jobConf, new Path(args[0]));
        FileOutputFormat.setOutputPath(jobConf, new Path(args[1]));

        jobConf.setOutputKeyClass(IntWritable.class);
        jobConf.setOutputValueClass(IntWritable.class);

        return jobConf.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new TestDriver(), args));
}

打包在jar中并使用'hadoop jar test.jar infile.txt / tmp / testout'运行 .

1 回答

  • 0

    Hadoop在流式传输reducer值时重用值对象 .

    因此,为了捕获所有不同的值,您需要复制:

    @Override
    protected void reduce(IntWritable key, Iterable<IntWritable> valueItr, Context context) throws  IOException, InterruptedException {        
        List<IntWritable> values = Lists.newArrayList();
        for(IntWritable writable : valueItr) {
            values.add(new IntWritable(writable.get());
        }
    
        System.err.println(key + "|" + values);
    }
    

相关问题