我使用hadoop的mapreduce从hdfs读入一个文件,通过一个简单的解析器将其放入,然后将解析器的输出写回hdfs。我还没有任务。我想知道为什么我的输出文件中有大约300个副本。
这是我的Map方法。
public void map(LongWritable key, Text value,
OutputCollector<Text, Text> output, Reporter reporter)
throws IOException {
FileSplit fsplit = (FileSplit) reporter.getInputSplit();
Main parser = new Main();
String datFilePath = fsplit.getPath().getName();
String valueMap = "/path/to/file";
Path pt = fsplit.getPath();
FileSystem fs = null;
try {
fs = FileSystem.get(new URI("hdfs://xxx.xxx.x.x:xxxx"),
new Configuration());
} catch (URISyntaxException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
try (FSDataInputStream inputStream = fs.open(pt)) {
ReadableByteChannel channel = Channels.newChannel(inputStream);
ByteBuffer buffer = ByteBuffer.allocate((int) fs.getFileStatus(pt).getLen());
channel.read(buffer);
buffer.order(ByteOrder.LITTLE_ENDIAN);
SimpleKeyValueStructure map = parser.parse(datFilePath, buffer,
valueMap);
String lrtransPath = map.getInputIdentifier();
SortedMap<String, Object> data = map.getData();
for (Entry<String, Object> entry : data.entrySet()) {
term.set(entry.getKey());
pathToFile.set(entry.getValue().toString());
output.collect(term, pathToFile);
}
count += 1;
System.out.println(count);
}
}
}
我最后打印出来的数字确实是3xx。这是配置问题吗?我的作业配置:
JobConf conf = new JobConf(MapReduce.class);
conf.setJobName("jobxyz");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(Text.class);
conf.setMapperClass(Map.class);
conf.setCombinerClass(Reduce.class);
conf.setReducerClass(Reduce.class);
conf.setNumReduceTasks(0);
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
FileInputFormat.setInputPaths(conf, new Path(args[0]));
FileOutputFormat.setOutputPath(conf, new Path(args[1]));
JobClient.runJob(conf);
输出完全正确,但重复。
1条答案
按热度按时间krugob8w1#
一
Mapper
为文件的每个输入分割调用;以及Mapper
的map()
对每个记录都调用。因为您的代码是针对每个输入分割中的每条记录运行的,所以您得到的是重复的。