MapReduce 二次排序：对表格数据进行多字段排序

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

例如有以下数据:

| first_name | last_name | age |
|---|---|---| | A | Smith | 25 | | B | Jones | 30 | | C | Brown | 22 | | B | White | 28 | | A | Black | 27 | | C | Green | 29 | | A | Davis | 24 |

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

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 之间用空格分隔，其他字段用英文分号分隔

内容:由于缺少数据，下面的代码是一个基本的二次排序的示例，实现了按照第一个字段升序排序，第二个字段降序排序的功能。

Mapper:

public class SortMapper extends Mapper<LongWritable, Text, MyKey, Text> {

    private MyKey keyInfo = new MyKey();
    private Text valueInfo = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] fields = value.toString().split(',');
        String first = fields[0];
        int second = Integer.parseInt(fields[1]);
        keyInfo.setFirst(first);
        keyInfo.setSecond(second);
        valueInfo.set(fields[1]);
        context.write(keyInfo, valueInfo);
    }
}

Reducer:

public class SortReducer extends Reducer<MyKey, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();

    @Override
    protected void reduce(MyKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        for (Text value : values) {
            outputKey.set(key.getFirst());
            outputValue.set(key.getSecond() + '');
            context.write(outputKey, outputValue);
        }
    }
}

MyKey:

public class MyKey implements WritableComparable<MyKey> {

    private String first;
    private int second;

    public MyKey() {
    }

    public MyKey(String first, int second) {
        this.first = first;
        this.second = second;
    }

    public String getFirst() {
        return first;
    }

    public void setFirst(String first) {
        this.first = first;
    }

    public int getSecond() {
        return second;
    }

    public void setSecond(int second) {
        this.second = second;
    }

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(first);
        dataOutput.writeInt(second);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        first = dataInput.readUTF();
        second = dataInput.readInt();
    }

    @Override
    public int compareTo(MyKey o) {
        int cmp = first.compareTo(o.getFirst());
        if (cmp != 0) {
            return cmp;
        }
        return o.getSecond() - second;
    }
}

Driver:

public class SortDriver {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, 'secondary sort');
        job.setJarByClass(SortDriver.class);
        job.setMapperClass(SortMapper.class);
        job.setReducerClass(SortReducer.class);
        job.setOutputKeyClass(MyKey.class);
        job.setOutputValueClass(Text.class);
        job.setPartitionerClass(FirstCharPartitioner.class);
        job.setGroupingComparatorClass(FirstCharGroupingComparator.class);
        job.setSortComparatorClass(MyKeyComparator.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

FirstCharPartitioner:

public class FirstCharPartitioner extends Partitioner<MyKey, Text> {

    @Override
    public int getPartition(MyKey key, Text value, int numPartitions) {
        return key.getFirst().toLowerCase().charAt(0) - 'a';
    }
}

FirstCharGroupingComparator:

public class FirstCharGroupingComparator extends WritableComparator {

    protected FirstCharGroupingComparator() {
        super(MyKey.class, true);
    }

    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        MyKey key1 = (MyKey) a;
        MyKey key2 = (MyKey) b;
        return key1.getFirst().compareToIgnoreCase(key2.getFirst());
    }
}

MyKeyComparator:

public class MyKeyComparator extends WritableComparator {

    protected MyKeyComparator() {
        super(MyKey.class, true);
    }

    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        MyKey key1 = (MyKey) a;
        MyKey key2 = (MyKey) b;
        return key1.compareTo(key2);
    }
}

代码说明:

MyKey: 自定义的 key 类，包含了两个排序字段 first 和 second，实现了 Comparable 接口，用于定义排序规则。
SortMapper: 将输入数据解析成 <MyKey, Text> 对，并将数据写入 context。
SortReducer: 将相同 key 的数据进行二次排序，并输出排序后的结果。
FirstCharPartitioner: 将数据根据第一个字段的首字母进行分区。
FirstCharGroupingComparator: 在每个分区内对数据进行分组，并将相同第一个字段的值进行比较。
MyKeyComparator: 对 MyKey 对象进行排序。

注意:

本代码仅供参考，需要根据实际情况进行调整。
需要添加依赖库：hadoop-core、commons-logging、commons-io 等。
可以使用 Maven 或其他工具构建项目。
需要将代码打包成 jar 文件，并使用 Hadoop 命令运行。