MapReduce中如何实现Reducer仅输出排序后的Top 5记录？

你猜的完全没错！问题核心就出在count变量的作用域上——把它放在reduce()方法内部的话，每次框架调用reduce()处理一条输入记录时，这个变量都会被重新初始化，自然没法累计计数。

要解决这个问题，你只需要把count提升为Reducer类的成员变量，而不是方法局部变量。这样每个Reducer实例会维护自己的count状态，不会在每次调用reduce()时重置。如果你的需求是每个键分组输出前5条，还可以在setup()方法里重置count，确保每个分组都重新计数。

具体代码实现示例 #

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class Top5Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    // 成员变量：每个Reducer实例独立维护这个count
    private int count;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // 每个键分组处理前重置count，保证每个分组都能取前5条
        count = 0;
    }

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        for (IntWritable value : values) {
            if (count < 5) {
                // 输出当前记录，然后计数+1
                context.write(key, value);
                count++;
            } else {
                // 已经取够5条，直接跳出循环停止处理
                break;
            }
        }
    }
}

额外注意事项 #

1. 排序规则要正确：确保Map阶段输出的键值对已经按照你需要的排名顺序排序（比如按值降序），这样Reducer拿到的前5条才是真正的Top5。你可以通过自定义WritableComparable或者配置Job的排序comparator来实现。
2. 全局Top5 vs 分组Top5：
   • 如果要的是全局前5，需要把Job的Reducer数量设置为1（job.setNumReduceTasks(1)），这样所有数据都会被同一个Reducer处理，count会累计全局的前5条。
   • 如果是每个键分组的前5，直接用上面的代码即可，每个分组都会独立计数前5。

内容的提问来源于stack exchange，提问作者Arpan Parikh