Я написал программу Map and Reduce, в которой ключ и значение Output Reducer отличаются от входных или выходных данных Mapper. Я сделал соответствующие изменения в классе водителя. Вот исключение, которое я получаю при его запуске:
INFO mapreduce.Job: идентификатор задачи: try_1550670375771_4211_m_000003_2, статус: не выполнено
Ошибка: java.io.IOException: несоответствие типа в значении из карты: ожидается org.apache.hadoop.io.Text, получено org.apache.hadoop.io.FloatWritable
в org.apache.hadoop.mapred.MapTask $ MapOutputBuffer.collect (MapTask.java:1084)
в org.apache.hadoop.mapred.MapTask $ NewOutputCollector.write (MapTask.java:721)
в org.apache.hadoop.mapreduce.task.TaskInputOutputContextImpl.write (TaskInputOutputContextImpl.java:89)
в org.apache.hadoop.mapreduce.lib.map.WrappedMapper $ Context.write (WrappedMapper.java:112)
в com.hirw.maxcloseprice.MyHadoopMapper.map (MyHadoopMapper.java:20)
в com.hirw.maxcloseprice.MyHadoopMapper.map (MyHadoopMapper.java:1)
в org.apache.hadoop.mapreduce.Mapper.run (Mapper.java:145)
в org.apache.hadoop.mapred.MapTask.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.doPrivileged (собственный метод)
в javax.security.auth.Subject.doAs (Subject.java:415)
в org.apache.hadoop.security.UserGroupInformation.doAs (UserGroupInformation.java:1920)
в org.apache.hadoop.mapred.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 не выполнено с состоянием FAILED из-за: Задача не выполнена task_1550670375771_4211_m_000001
Задание не выполнено, поскольку задание не выполнено. failedMaps: 1 failedReduces: 0
Работает, когда KeyOut и ValueOut в Reducer такие же, как в Mapper, но не работает, когда они отличаются.
Мой класс 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));
}
}
Класс редуктора:
открытый класс MyHadoopReducer расширяет Редуктор {
@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)));
}
}
и класс водителя:
открытый класс 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);
}
}