假设在某一个目录下有若干文本文件，现在需要对文件中的数据进行二次排序，即首先根据第一列数据降序排序，如果第一列数据相等需要根据第二列数据降序排序具体步骤如下1)定义一个类SecondarySortKey，实现自定义的用于排序的key2)将需要进行二次排序的文件加载进来生成(key,value)键值对类型的RDD3)使用sortByKey()基于自定义的key进行二次排序4)去掉排序的key，只保留排序的结果

代码如下：

1. 定义SecondarySortKey类：

import java.io.Serializable;
import scala.math.Ordered;

public class SecondarySortKey implements Ordered<SecondarySortKey>, Serializable {
    private static final long serialVersionUID = 1L;
    private int first;
    private int second;
    
    public SecondarySortKey(int first, int second) {
        this.first = first;
        this.second = second;
    }

    public int getFirst() {
        return first;
    }

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

    public int getSecond() {
        return second;
    }

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

    @Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        }
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        SecondarySortKey other = (SecondarySortKey) obj;
        if (first != other.first) {
            return false;
        }
        if (second != other.second) {
            return false;
        }
        return true;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + first;
        result = prime * result + second;
        return result;
    }

    @Override
    public boolean $less(SecondarySortKey other) {
        if (first < other.first) {
            return true;
        } else if (first == other.first && second < other.second) {
            return true;
        }
        return false;
    }

    @Override
    public boolean $greater(SecondarySortKey other) {
        if (first > other.first) {
            return true;
        } else if (first == other.first && second > other.second) {
            return true;
        }
        return false;
    }

    @Override
    public boolean $less$eq(SecondarySortKey other) {
        if (first < other.first) {
            return true;
        } else if (first == other.first && second <= other.second) {
            return true;
        }
        return false;
    }

    @Override
    public boolean $greater$eq(SecondarySortKey other) {
        if (first > other.first) {
            return true;
        } else if (first == other.first && second >= other.second) {
            return true;
        }
        return false;
    }

    @Override
    public int compare(SecondarySortKey other) {
        if (first - other.first != 0) {
            return first - other.first;
        } else {
            return second - other.second;
        }
    }
}

2. 加载文件并生成(key,value)键值对类型的RDD：

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

public class SecondarySortApp {
    public static void main(String[] args) {
        JavaSparkContext sc = new JavaSparkContext("local", "SecondarySortApp");
        JavaRDD<String> lines = sc.textFile("file:///path/to/files/*");
        JavaPairRDD<SecondarySortKey, String> pairs = lines.mapToPair(line -> {
            String[] tokens = line.split(",");
            int first = Integer.parseInt(tokens[0]);
            int second = Integer.parseInt(tokens[1]);
            return new Tuple2<>(new SecondarySortKey(first, second), line);
        });
    }
}

3. 使用sortByKey()基于自定义的key进行二次排序：

JavaPairRDD<SecondarySortKey, String> sortedPairs = pairs.sortByKey(false);

4. 去掉排序的key，只保留排序的结果：

JavaRDD<String> sortedLines = sortedPairs.map(tuple -> tuple._2);
sortedLines.saveAsTextFile("file:///path/to/output");