Hadoop Map Reduce，如何将第一个reducer输出和第一个map输入结合起来，作为第二个mapper的输入？

Question

我需要使用 map reduce 实现一个功能。

要求如下。

1. 映射器的输入是一个包含两列 productId 和 Salescount 的文件
2. Reducers output , salescount 总和

要求是我需要计算 salescount/sum(salescount)。

为此，我打算使用嵌套的 map reduce。但是对于第二个映射器，我需要使用第一个 reducer 输出和第一个映射的输入。

我怎样才能实现这个。或者有什么替代方法吗？

问候维努

Answer 1

您可以按照自己的方式使用 ChainMapper 和 ChainReducer 到 PIPE Mappers 和 Reducer。请看here

以下将类似于您需要实现的代码片段

JobConf mapBConf = new JobConf(false);

JobConf reduceConf = new JobConf(false);

ChainMapper.addMapper(conf, FirstMapper.class, FirstMapperInputKey.class, FirstMapperInputValue.class,
   FirstMapperOutputKey.class, FirstMapperOutputValue.class, false, mapBConf);

ChainReducer.setReducer(conf, FirstReducer.class, FirstMapperOutputKey.class, FirstMapperOutputValue.class,
   FirstReducerOutputKey.class, FirstReducerOutputValue.class, true, reduceConf);

ChainReducer.addMapper(conf, SecondMapper.class, FirstReducerOutputKey.class, FirstReducerOutputValue.class,
   SecondMapperOutputKey.class, SecondMapperOutputValue.class, false, null);

ChainReducer.setReducer(conf, SecondReducer.class, SecondMapperOutputKey.class, SecondMapperOutputValue.class, SecondReducerOutputKey.class, SecondReducerOutputValue.class, true, reduceConf);

或者如果您不想使用多个 Mappers 和 Reducers，您可以执行以下操作

public static class ProductIndexerMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> {

    private static Text productId = new Text();
    private static LongWritable salesCount = new LongWritable();

    @Override
    public void map(LongWritable key, Text value,
            OutputCollector<Text, LongWritable> output, Reporter reporter)
            throws IOException {
        String[] values = value.toString().split("\t");
        productId.set(values[0]);           
        salesCount.set(Long.parseLong(values[1]));
        output.collect(productId, salesCount);
    }

}

public static class ProductIndexerReducer extends MapReduceBase implements Reducer<Text, LongWritable, Text, LongWritable> {

    private static LongWritable productWritable = new LongWritable();

    @Override
    public void reduce(Text key, Iterator<LongWritable> values,
            OutputCollector<Text, LongWritable> output, Reporter reporter)
            throws IOException {
        List<LongWritable> items = new ArrayList<LongWritable>(); 
        long total = 0;
        LongWritable item = null;
        while(values.hasNext()) {
            item = values.next();
            total += item.get();
            items.add(item);
        }
        Iterator<LongWritable> newValues = items.iterator();
        while(newValues.hasNext()) {
            productWritable.set(newValues.next().get()/total);
            output.collect(key, productWritable);
        }
    }

}

`