Hadoop MapReduce词频统计全局排序输出为空问题求助

嘿，我之前在优化Hadoop WordCount的时候也踩过TotalOrderPartitioner导致输出为空的坑，结合你的场景（加了Combiner+全局排序需求），大概率是这几个核心原因，给你拆解下：

一、最常见的原因：没配置分区边界文件 #

TotalOrderPartitioner和默认的HashPartitioner不一样，它依赖一个提前生成的分区边界文件来划分每个Reducer要处理的Key范围。如果没生成这个文件，或者没告诉Job它的位置，Reducer根本接收不到数据，自然输出为空。

二、Combiner与Reducer的类型不匹配（隐性坑） #

虽然Combiner的输出理论上要和Reducer输入一致，但如果你的Combiner不小心改了Key的类型（比如WordCount里误把Text改成了别的），而分区边界文件是基于原始Mapper的Key生成的，就会导致分区逻辑完全失效，数据无法分发到正确的Reducer。

三、Reducer数量和分区数不匹配 #

TotalOrderPartitioner的分区数必须和你设置的job.setNumReduceTasks(n)的n完全一致。比如你生成了5个分区，但只启动了3个Reducer，那后面2个分区的数据找不到对应的Reducer，直接丢失；反过来，Reducer多了的话，部分Reducer会空跑。

解决办法（附完整代码片段） #

我把修复后的关键代码整理出来，你可以对照调整：

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;

public class OptimizedWordCount {

    // Mapper类（和官方示例一致即可）
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }

    // 你的Combiner类（确保输出类型和Reducer输入一致）
    public static class WordCountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

    // Reducer类（和官方示例一致即可）
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "optimized-wordcount-with-global-sort");

        // 基础配置
        job.setJarByClass(OptimizedWordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(WordCountCombiner.class); // 启用Combiner
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 核心：配置TotalOrderPartitioner
        job.setPartitionerClass(TotalOrderPartitioner.class);
        // 生成分区边界文件：用随机采样，参数分别是采样比例、最大采样数、采样次数
        InputSampler.Sampler<Text, IntWritable> sampler = new InputSampler.RandomSampler<>(0.2, 1000, 5);
        // 指定分区文件的HDFS路径（也可以用临时路径）
        Path partitionFile = new Path("/tmp/wordcount-partition-boundaries.lst");
        InputSampler.writePartitionFile(job, sampler, partitionFile);
        // 告诉TotalOrderPartitioner分区文件的位置
        conf.set(TotalOrderPartitioner.PARTITION_FILE, partitionFile.toString());

        // 设置Reducer数量，必须和分区数一致（默认采样生成的分区数等于Reducer数）
        job.setNumReduceTasks(3); // 按需调整，比如设为1就是单文件全局排序，但大数据量下效率低

        // 输入输出路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

额外检查项

• 确认HDFS上的分区文件路径有读写权限，Job运行用户能访问
• 如果是本地模式测试，注意分区文件路径是本地文件系统还是HDFS（本地模式下可以用file:///前缀指定本地路径）
• 如果你是想按词频排序（而不是单词字典序），那需要把Key和Value互换，并用TotalOrderPartitioner基于IntWritable排序，还要处理相同词频的二次排序，这个场景可以再调整代码

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