Hadoop MapReduce 数据排序实战：正序与逆序实现

2. 降序排序

如果需要逆序排列，只需自定义比较器即可。MapReduce 允许通过 setSortComparatorClass 指定 Key 的比较规则。这里我们实现 RawComparator，反转整数的比较逻辑。

package com.mk.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
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 java.io.IOException;
import java.net.URI;

public class Sort {
    public static class SortMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            IntWritable v = new IntWritable(Integer.parseInt(value.toString().trim()));
            context.write(v, new IntWritable(1));
        }
    }

    public static class SortReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
        int count = 1;

        @Override
        protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            for (IntWritable v : values) {
                context.write(new IntWritable(count++), key);
            }
        }
    }

    // 自定义比较器实现降序
    public static class SortComparator implements RawComparator<IntWritable> {
        @Override
        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
            return IntWritable.Comparator.compareBytes(b2, s2, l2, b1, s1, l1);
        }

        @Override
        public int compare(IntWritable o1, IntWritable o2) {
            return o2.get() - o1.get();
        }
    }

    public static void main(String[] args) throws Exception {
        String uri = "hdfs://192.168.150.128:9000";
        String input = "/sort/input";
        String output = "/sort/output";

        Configuration conf = new Configuration();
        if (System.getProperty("os.name").toLowerCase().contains("win")) {
            conf.set("mapreduce.app-submission.cross-platform", "true");
        }

        FileSystem fileSystem = FileSystem.get(URI.create(uri), conf);
        Path path = new Path(output);
        fileSystem.delete(path, true);

        Job job = new Job(conf, "Sort");
        job.setJar("./out/artifacts/hadoop_test_jar/hadoop-test.jar");
        job.setJarByClass(Sort.class);
        job.setMapperClass(SortMapper.class);
        job.setReducerClass(SortReducer.class);

        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPaths(job, uri + input);
        FileOutputFormat.setOutputPath(job, new Path(uri + output));

        // 关键：设置自定义比较器
        job.setSortComparatorClass(SortComparator.class);

        boolean ret = job.waitForCompletion(true);
        System.out.println(job.getJobName() + "-----" + ret);
    }
}

注意事项

1. 路径配置：代码中的 HDFS URI 和本地 Jar 包路径需要根据实际环境修改，尤其是 job.setJar 部分，确保指向编译后的真实 JAR 文件。
2. 数据类型：使用 IntWritable 而非 Text 是关键，否则排序会变成字符串字典序（例如 "10" 会排在 "2" 前面）。
3. Combiner：虽然本例未使用，但在数据量极大时，添加 Combiner 可以减少网络传输开销，不过需注意 Combiner 不能改变最终输出的顺序逻辑。