Spark的二次排序

Posted 指尖上的艺术

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark的二次排序相关的知识,希望对你有一定的参考价值。

1、数据样本:

1 5
2 4
3 6
1 3
2 1
1 14
2 45
4 11
3 23
5 12
6 13

2、排序规则:先按照第一个字符排序,如果第一个相同,再按照第二个字符排序

3、排序后的结果

1 3
1 5
1 14
2 1
2 4
2 45
3 6
3 23
4 11
5 12
6 13

4、spark二次排序实现

 4.1、自定义key

package com.test.spark

/**
  * @author admin
  * scala处理二次排序的类
  * 自定义key
  */
class SecondSortByKey(val first: Int, val second: Int) extends Ordered[SecondSortByKey] with Serializable {
  def compare(other: SecondSortByKey): Int = {
    //this关键字可加,也可不加,如果遇到多个变量时,必须添加
    if (this.first - other.first != 0)
      this.first - other.first
    else
      this.second - other.second
  }

  //重写toString方法
  /*override def toString(): String = {
    "first:" + first + " second:" + second
  }*/
}

 

4.2、二次排序程序编排

package com.test.spark

import org.apache.spark.{SparkConf, SparkContext}

/**
  * @author admin
  * Spark二次排序的具体实现步骤:
  * 第一步: 自定义key 实现scala.math.Ordered接口,和Serializeable接口
  * 第二步:将要进行二次排序的数据加载,按照<key,value>格式的RDD
  * 第三步:使用sortByKey 基于自定义的key进行二次排序
  * 第四步:去掉排序的key,只保留排序的结果
  */
object SparkSecondSortApplication {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SortSecond").setMaster("local[1]")
    // 获取context
    val sc = new SparkContext(conf)
    // 加载到内存RDD
    val lines = sc.textFile("D:\\SparkDataTest\\sort.txt")
    // map操作,将要进行二次排序的数据加载,按照<key,value>格式的RDD
    val pairs = lines.map { line => {
      val spl = line.split(" ")
      (new SecondSortByKey(spl(0).toInt, spl(1).toInt), line)
    }
    }
    // 使用sortByKey 基于自定义的key进行二次排序, true:升序,false:降序
    val sortPair = pairs.sortByKey(true)

    // map操作,只需要保留排序结果
    val sortResult = sortPair.map(line => line._2)

    sortResult.collect().foreach { x => println(x) }

    // 停止sc
    sc.stop()
  }
}

 

以上是关于Spark的二次排序的主要内容,如果未能解决你的问题,请参考以下文章

Spark 中的二次排序Java实现

Spark 使用sortByKey进行二次排序

mapreduce 的二次排序

基于spark SQL之上的检索与排序对比性能测试

ExtJS4.1 TreeGrid 中的二次排序

问题排查 | Spark OrcFileFormat inferSchema执行巨慢问题分析