为什么我在Hadoop的mapreduce中得到3xx个重复项？

Question

我使用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);

输出完全正确，但是重复。

Answer 1

每个文件的输入分割都会调用一个Mapper;每个记录都调用Mapper的map()。由于您的代码在每个输入分割中运行每个记录，因此您将获得重复记录。