MapReduce 二次排序实现数据排序

MapReduce 在传递<key, value>对时默认根据 key 进行排序，所以当对表格数据按某一字段进行排序时可以设置排序的字段为 key，而有时候我们需要对两个字段分别进行排序，基于这种需求进行的自定义排序称为'二次排序'。

例如有以下数据:

如果对上面的数据按第一个字段值进行升序排列，那么有些数据的顺序是相同的，这时候我们就要按照第二个字段值再进行一次排序，例如:先按照第一个值(字母)升序排列，再按照第二个值(数字)进行降序排列得到:

A, 3 B, 5 C, 1 B, 6 A, 4 C, 5 A, 4 A, 3 B, 6 B, 5 C, 5 C, 1

要求:编写一个 Java 程序，利用 MapReduce 技术对数据进行二次排序，把程序打包成 jar 文件运行得到结果。

(1)在程序中读取给定的表格文件 datas.xlsx 的内容(表格的第一行为标题，处理时需要排除)，首先对 first_name 按首字母(不区分大小写)进行分区，首字母相同的行在同一分区(即输出在同一

个文件，如 A、a 开头的行在第 1 个分区 part-r-00000，Z、z 开头的行在第 26 个分区 part-r-00025)

(2)然后在每个分区内对数据进行二次排序，要求先对 first_name 字段降序排列，再对 last_name 字段也进行降序排列，最后 MapReduce 输出的每一行结果中，first_name 和 last_name 之间用空格分隔，其他字段用英文分号分隔

内容：由于没有提供样例数据，以下程序只提供实现思路和大致代码框架。

实现思路：

1. 读取表格文件 datas.xlsx，排除第一行标题，获取 <first_name, last_name, ...> 的数据。
2. 自定义 Mapper，在 map 函数中将 first_name 作为 key，将 <first_name, last_name, ...> 作为 value 输出。
3. 自定义 Partitioner，按照 key 的首字母进行分区。
4. 自定义 KeyComparator，按照 first_name 字段进行降序排列。
5. 自定义 GroupComparator，按照 key 进行分组。
6. 自定义 ValueComparator，按照 last_name 字段进行降序排列。
7. 自定义 Reducer，在 reduce 函数中将 value 按照 last_name 字段进行排序后输出。

代码框架：

// 自定义 Mapper public static class MyMapper extends Mapper<LongWritable, Text, Text, Text> { public void map(LongWritable key, Text value, Context context) { // 读取表格文件 datas.xlsx，排除第一行标题，获取 <first_name, last_name, ...> 的数据 // 将 first_name 作为 key，将 <first_name, last_name, ...> 作为 value 输出 context.write(new Text(first_name), new Text(value.toString())); } }

// 自定义 Partitioner public static class MyPartitioner extends Partitioner<Text, Text> { public int getPartition(Text key, Text value, int numReduceTasks) { // 按照 key 的首字母进行分区 return (int)key.charAt(0) - 65; } }

// 自定义 KeyComparator public static class MyKeyComparator extends WritableComparator { protected MyKeyComparator() { super(Text.class, true); } public int compare(WritableComparable a, WritableComparable b) { // 按照 first_name 字段进行降序排列 return -1 * ((Text)a).compareTo((Text)b); } }

// 自定义 GroupComparator public static class MyGroupComparator extends WritableComparator { protected MyGroupComparator() { super(Text.class, true); } public int compare(WritableComparable a, WritableComparable b) { // 按照 key 进行分组 return ((Text)a).compareTo((Text)b); } }

// 自定义 ValueComparator public static class MyValueComparator extends WritableComparator { protected MyValueComparator() { super(Text.class, true); } public int compare(WritableComparable a, WritableComparable b) { // 按照 last_name 字段进行降序排列 return -1 * ((Text)a).compareTo((Text)b); } }

// 自定义 Reducer public static class MyReducer extends Reducer<Text, Text, Text, Text> { public void reduce(Text key, Iterable values, Context context) { List list = new ArrayList<>(); for (Text value : values) { list.add(new Text(value.toString())); } // 将 value 按照 last_name 字段进行排序 Collections.sort(list, new Comparator() { public int compare(Text a, Text b) { return -1 * a.toString().split(",")[1].compareTo(b.toString().split(",")[1]); } }); for (Text value : list) { // 输出结果，first_name 和 last_name 之间用空格分隔，其他字段用英文分号分隔 context.write(key, new Text(value.toString().replace(",", " ").replaceFirst(" ", ";"))); } } }

// 主函数 public static void main(String[] args) throws Exception { // 配置作业 Job job = Job.getInstance(); job.setJarByClass(MyClass.class); job.setMapperClass(MyMapper.class); job.setReducerClass(MyReducer.class); job.setPartitionerClass(MyPartitioner.class); job.setSortComparatorClass(MyKeyComparator.class); job.setGroupingComparatorClass(MyGroupComparator.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); // 设置输入和输出路径 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 提交作业 System.exit(job.waitForCompletion(true) ? 0 : 1); }