java - 为什么我在hadoop的mapreduce中得到3xx重复？

我正在使用hadoop的mapreduce从hdfs中读取文件，将其通过一个简单的解析器，然后将该解析器的输出写回到hdfs。我还没有 reduce task 。我想知道为什么我的输出文件中有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);

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

最佳答案

文件的每个输入拆分都会调用Mapper；每个记录都会调用Mapper的map()。由于您的代码针对每个输入拆分中的每个记录运行，因此您将得到重复的代码。