MapReduce Reducer的KeyOut类型

Posted 2021-04-20

我编写了Map and Reduce程序，其中Reducer的Output键和值与输入或Mapper的输出不同。我在Driver的课上做了相应的更改。以下是我运行时遇到的异常：

INFO mapreduce.Job：任务ID：attempt_1550670375771_4211_m_000003_2，状态：FAILED错误：java.io.IOException：输入map中的值不匹配：expected org.apache.hadoop.io.Text，收到org.apache.hadoop.io.FloatWritable at org.apache.hadoop.mapred.MapTask $ MapOutputBuffer.collect（MapTask.java:1084）org.apache.hadoop.mapred.MapTask $ NewOutputCollector.write（MapTask.java:721）org.apache.hadoop.mapreduce。在com.hirw.maxcloseprice.MyHadoopMapper.map（MyHadoopMapper。）的org.apache.hadoop.mapreduce.lib.map.WrappedMapper $ Context.write（WrappedMapper.java:112）上的task.TaskInputOutputContextImpl.write（TaskInputOutputContextImpl.java:89）。 java：20）位于org.apache.hadoop.mapred.MapTask的org.apache.hadoop.mapreduce.Mapper.run（Mapper.java:145）的com.hirw.maxcloseprice.MyHadoopMapper.map（MyHadoopMapper.java:1） .runNewMapper（MapTask.java:793）位于org.apache.hadoop.mapred.MapTask.run（MapTask.java:341）org.apache.hadoop.mapred.YarnChild $ 2.run（YarnChild.java:164）at java .security.AccessController.d orP.vache.hadoop.mapred的org.apache.hadoop.security.UserGroupInformation.doAs（UserGroupInformation.java:1920）的javax.security.auth.Subject.doAs（Subject.java:415）上的特权（Native Method）。 YarnChild.main（YarnChild.java:158）

19/04/16 22:24:50 INFO mapreduce.Job：地图100％减少100％19/04/16 22:24:50 INFO mapreduce.Job：工作job_1550670375771_4211因状态失败而失败，原因是：任务失败task_1550670375771_4211_m_000001工作失败任务失败。 failedMaps：1次失败减少：0

当Reducer的KeyOut和ValueOut与Mapper的KeyOut和ValueOut相同但在它们不同时失败时，它可以工作。

我的Mapper类：公共类MyHadoopMapper扩展Mapper {

@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

    String[] recordItems = value.toString().split(",");

    String stock = recordItems[1];
    Float stockValue = Float.parseFloat(recordItems[6]);

    context.write(new Text(stock), new FloatWritable(stockValue));
}

}

Reducer类：

公共类MyHadoopReducer扩展Reducer {

@Override
public void reduce(Text key, Iterable<FloatWritable> values, Context context
        ) throws IOException, InterruptedException {

    Float maxVal = Float.MIN_VALUE;
    for (FloatWritable stockValue : values) {
        maxVal = stockValue.get() > maxVal ? stockValue.get() : maxVal;
    }

    context.write(key, new Text(String.valueOf(maxVal)));
}

}

和Driver类：公共类MyHadoopDriver {

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    // TODO Auto-generated method stub

    Job hadoopJob = new Job();
    hadoopJob.setJarByClass(MyHadoopDriver.class);
    hadoopJob.setJobName("MyStockPrice");

    FileInputFormat.addInputPath(hadoopJob, new Path("/user/hirw/input/stocks"));
    FileOutputFormat.setOutputPath(hadoopJob, new Path("stocksData"));

    hadoopJob.setInputFormatClass(TextInputFormat.class);
    hadoopJob.setOutputFormatClass(TextOutputFormat.class);

    hadoopJob.setMapperClass(MyHadoopMapper.class);
    hadoopJob.setReducerClass(MyHadoopReducer.class);

    hadoopJob.setCombinerClass(MyHadoopReducer.class);

    hadoopJob.setOutputKeyClass(Text.class);
    hadoopJob.setOutputValueClass(Text.class);

    System.exit(hadoopJob.waitForCompletion(true) ? 0: 1);
}

}

答案

默认情况下，映射器输出类型为Text，而使用FloatWritable。这就是异常告诉你的。您需要指定mapper输出类型，如下所示：

job.setMapOutputValueClass(FloatWritable.class)

另一答案

取出组合器或用匹配的键和输出写一个新的。