Spark实现二次排序

一.代码实现

package big.data.analyse.scala.secondsort

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

/** 二次排序
  * Created by zhen on 2019/5/29.
  */
class SecondSortByKey(val first:Int,val second:Int) extends Ordered[SecondSortByKey] with Serializable{
  def compare(other : SecondSortByKey): Int ={
    if(this.first - other.first != 0){//正序
      this.first - other.first
    }else{//倒序
      other.second - this.second
    }
  }
}
object SecondSortByKey{
  /**
    * 设置日志级别
    */
  Logger.getLogger("org").setLevel(Level.WARN)
  def main(args: Array[String]) {
    val spark = SparkSession
      .builder()
      .appName("SecondSortByKey")
      .master("local[2]")
      .getOrCreate()
    val sc = spark.sparkContext
    val rows = sc.textFile("src/big/data/analyse/scala/secondsort/sort.txt")

    val pairWithSortByKey = rows
      .filter(row=>row.split(" ").length==3)//过滤错误的数据
      .map(row=>{
        val array = row.split(" ")
        (new SecondSortByKey(array(0).toInt,array(1).toInt),row)
    })
    println("先正序后倒序")
    pairWithSortByKey
      .sortByKey(true) // 排序,true：先正序后倒序，false：先倒序后正序
      .map(map => map._2)
      .collect()
      .foreach(println)
    println("先倒序后正序")
    pairWithSortByKey
      .sortByKey(false) // 排序,true：先正序后倒序，false：先倒序后正序
      .map(map => map._2)
      .collect()
      .foreach(println)

    sc.stop()
  }
}

二.结果

先正序后倒序
1 9 ES
1 8 HBase
2 4 Tachyon日渐成熟
2 3 《黑豹》异军突起
2 3 Radis
3 3 HDFS
3 3 搜索引擎
5 3 spark发布2.4版本，性能提升巨大
5 2 《复仇者联盟3:无限战争》火热上映
6 3 Maven
7 2 Solr
先倒序后正序
7 2 Solr
6 3 Maven
5 2 《复仇者联盟3:无限战争》火热上映
5 3 spark发布2.4版本，性能提升巨大
3 3 HDFS
3 3 搜索引擎
2 3 《黑豹》异军突起
2 3 Radis
2 4 Tachyon日渐成熟
1 8 HBase
1 9 ES

Process finished with exit code 0