Hadoop MapReduce：多文件词频统计结果按文件单行输出求助

你遇到的核心问题是Reducer把不同文件中相同词汇的词频错误求和了，而且没能按文件维度分组输出。这是因为你的Map阶段输出的Key只包含词汇本身，Reducer无法区分同一个词来自哪个文件，自然会把所有文件的统计值合并，最后每个文件的输出都是全量总和的重复。

下面是具体的修复方案，分步骤来：

1. 修改Map阶段：让Key关联文件名 #

在Map任务中，你需要获取当前处理的文件名，把文件名+词汇作为输出的Key，这样Reducer就能区分不同文件的同一词汇。

示例代码片段：

public class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text keyOutput = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 获取当前处理的文件名
        FileSplit split = (FileSplit) context.getInputSplit();
        String fileName = split.getPath().getName();
        
        // 分割每行的词汇
        String[] words = value.toString().split("\\s+");
        for (String word : words) {
            // 构造Key：文件名+分隔符+词汇，比如"file1:hello"
            keyOutput.set(fileName + ":" + word);
            context.write(keyOutput, one);
        }
    }
}

2. 调整Reducer阶段：按文件名分组整理输出 #

Reducer需要先按文件名分组，把同一个文件下的所有词汇和词频收集起来，最后一次性输出一行该文件的统计结果。

示例代码片段：

public class MyReducer extends Reducer<Text, IntWritable, Text, Text> {
    private Text resultValue = new Text();
    // 全局维护文件名到词频映射的Map
    private Map<String, StringBuilder> fileWordCounts = new HashMap<>();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        String[] parts = key.toString().split(":");
        String fileName = parts[0];
        String word = parts[1];
        
        // 计算当前词汇的总词频
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        
        // 把当前词频对添加到对应文件的StringBuilder中
        fileWordCounts.computeIfAbsent(fileName, k -> new StringBuilder()).append(word).append(":").append(sum).append(", ");
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 遍历所有文件的统计结果，逐行输出
        for (Map.Entry<String, StringBuilder> entry : fileWordCounts.entrySet()) {
            String fileName = entry.getKey();
            // 移除最后多余的", "
            String counts = entry.getValue().toString().replaceAll(", $", "");
            resultValue.set(counts);
            context.write(new Text(fileName), resultValue);
        }
    }
}

3. 确保输出为单个文件 #

为了让结果输出到一个文件中，需要在Job配置中设置Reducer任务数为1（适合数据量不大的场景）：

job.setNumReduceTasks(1);

为什么原来的代码会出错？ #

你之前的Map阶段只输出词汇作为Key，不管它来自哪个文件。Reducer收到所有文件中同一词汇的IntWritable值后，会默认把它们求和，最后每个文件的输出其实是所有文件的词频总和，自然会出现重复的错误结果。

内容的提问来源于stack exchange，提问作者Amy Chan