我在理解如何将简单hadoop的输出写回elasticsearch时遇到了一些困难。
作业配置为:
job.setOutputFormatClass(EsOutputFormat.class);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(MapWritable.class);
减速机有:
final DoubleWritable average = new DoubleWritable(sum / size);
final MapWritable output = new MapWritable();
output.put(key, average);
context.write(NullWritable.get(), output);
但我有一个莫名其妙的例外:
14/08/15 16:59:54 INFO mapreduce.Job: Task Id : attempt_1408106733881_0013_r_000000_2, Status : FAILED Error:
org.elasticsearch.hadoop.EsHadoopIllegalArgumentException:
[org.elasticsearch.hadoop.serialization.field.MapWritableFieldExtractor@5796fabe] cannot extract value from object [org.apache.hadoop.io.MapWritable@dcdb8e97]
at org.elasticsearch.hadoop.serialization.bulk.TemplatedBulk$FieldWriter.write(TemplatedBulk.java:49)
at org.elasticsearch.hadoop.serialization.bulk.TemplatedBulk.writeTemplate(TemplatedBulk.java:101)
at org.elasticsearch.hadoop.serialization.bulk.TemplatedBulk.write(TemplatedBulk.java:77)
at org.elasticsearch.hadoop.rest.RestRepository.writeToIndex(RestRepository.java:130)
at org.elasticsearch.hadoop.mr.EsOutputFormat$EsRecordWriter.write(EsOutputFormat.java:161)
at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.write(ReduceTask.java:558)
在context.write()调用中。
我有点困惑,知道吗?
1条答案
按热度按时间gv8xihay1#
原来我在工作配置上犯了个错误。我补充了以下内容:
没有在输出mapwritable中实际添加\u id字段;这导致es抛出异常。
如果mapwritablefleldextractor记录了它失败的字段,这将非常有用。