Hadoop MapReduce 数据排序实战
数据排序是数据处理流程中的基础环节，无论是学生成绩排名还是建立索引，往往都需要先对原始数据进行有序化处理。这个案例的逻辑与数据去重类似，都是通过对输入数据的初步清洗和重组，为后续操作铺平道路。
需求分析
我们需要处理一个包含多行数字的输入文件。每行代表一条数据，目标是在输出中生成两列：第一列是该数字在整体排序后的位次（Rank），第二列是原始数值本身。
假设输入包含三个文件，内容如下：
file1
2 32 654 32 15 756 65223
file2
5956 22 650 92
file3
26 54 6
期望的输出结果应体现全局排序后的序号与数值的对应关系：
1 2
2 6
3 15
...
14 65223
核心思路
MapReduce 框架本身具备强大的排序能力，默认会根据 Key 的值进行排序。如果我们把要排序的数字封装成 IntWritable 作为 Key 输出，框架就会自动按数字大小排列；如果是 Text 类型，则按字典序排列。
在这个场景中，我们采用 IntWritable 作为 Key。Mapper 阶段将读取到的数字转换为整数并写入 Context，Value 可以设为任意占位符（例如 1）。Reducer 接收到 <Key, ValueList> 后，利用 ValueList 的长度判断该数字出现的次数，同时维护一个全局计数器来生成位次。
需要注意的是，本示例未配置 Combiner。虽然 Combiner 能优化性能，但在这种简单的计数排序场景下，直接使用 Mapper 和 Reducer 足以完成任务，且逻辑更清晰。
代码实现
1. 升序排序
升序是 MapReduce 的默认行为，关键在于正确设置 Key 的类型。Mapper 负责解析文本，Reducer 负责分配序号。
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.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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 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，作为 Key 输出
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 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));
boolean ret = job.waitForCompletion(true);
System.out.println(job.getJobName() + "-----" + ret);
}
}
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);
}
}
注意事项
- 路径配置：代码中的 HDFS URI 和本地 Jar 包路径需要根据实际环境修改，尤其是
job.setJar部分，确保指向编译后的真实 JAR 文件。 - 数据类型：使用
IntWritable而非Text是关键，否则排序会变成字符串字典序（例如 "10" 会排在 "2" 前面）。 - Combiner：虽然本例未使用，但在数据量极大时，添加 Combiner 可以减少网络传输开销，不过需注意 Combiner 不能改变最终输出的顺序逻辑。