自定义排序

例题：

先按照数学成绩进行倒叙排序  如果数学成绩相同 语文成绩谁高谁在前
tom 98 66
jack 55 55
bob 98 44
joe 44 99
max 60 79
peter 60 55
jerry 60 99
kay 99 99
kim 98 99

代码实现 带*****号表示是 关键词

object Z_Stor extends App {
     val conf = new SparkConf()
       .setAppName("sort")
       .setMaster("local[2]")
       val sc = new SparkContext(conf)
  sc.setLogLevel("ERROR")
  //从桌面读取数据
  val line = sc.textFile("C:\\Users\\Undo\\Desktop\\sort.txt").map(l=>{
    val lin=l.split(" ")
    (lin(0),lin(1).toInt,lin(2).toInt)
  })
//调用自己定义的排序类   ******调用sortBy() 方法         ******* false表示最后结果是倒序排序
  val order = line.sortBy(x=>UDFSort(x._2,x._3),false)
println(order.collect().toBuffer)   //***************foreach（）输出会出现 乱序

sc.stop()//关流
}
//排序的样例类   *******继承Ordered
case class UDFSort(val cn:Int,val math:Int)extends Ordered[UDFSort]{
  override def compare(that: UDFSort): Int = {  //排序方法
    //先按照数学成绩进行倒叙排序  如果数学成绩相同 语文成绩谁高谁在前
    if(this.math==that.math){
      this.cn-that.cn
    }else{
      this.math-that.math   //传过来的减去之前的（现在的跟之前的进行比较）
    }

  }
}

正确结果

自定义分区 然后再各自分区内倒序排序，取前两位

例题 做题所需文档在百度网盘上 链接：

链接：https://pan.baidu.com/s/1w9e1soXjpNDCLXEYx3_Eow
提取码：9wwh

字段说明 http://bigdata.baway.cn/lucas中 bigdata 为学科，lucas为老师
使用sparksql来进行统计每个学科最受欢迎的老师Top2（5）

代码

将文件读取路径（args(0)）设置成活动的

package com.day_214

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

import scala.collection.mutable

/**
 * TODO
 *
 * @author 徐磊
 * @email [email protected]
 * @data2020/02/14 下午 06:28
 * @最终需求效果： C:\Users\Undo\Desktop\tacher.log
 *          字段说明 http://bigdata.baway.cn/lucas中 bigdata 为学科，lucas为老师
 *          使用sparksql来进行统计每个学科里面最受欢迎的老师Top2（5）
 *  思路：1、先将不同的学科放到不同的文档
 *       2、对每一个文档进行倒序排序，最后取出前两位
 */
object Z_Fq extends App {
  val con = new SparkConf()
    .setAppName("ZFq")
    .setMaster("local[2]")
  val sc = new SparkContext(con)
  sc.setLogLevel("ERROR")
  val TopN=args(1).toInt    //TopN也是从右上角设置的
  //读取数据   形成（学科，老师）的  key-value 形式
  val res = sc.textFile(args(0)).map(line=>{
    val xk=line.substring(line.indexOf("//")+2,line.indexOf("."))
    val teach=line.substring(line.lastIndexOf("/")+1,line.length)
    ((xk,teach),1)
  })

  //分区之前需要把每个老师出现的次数求出来
  val reducerres=res.reduceByKey(_+_)
  //获取老师名字  以确定接下来要设置几个分区
val teachname=reducerres.map(_._1._1).distinct().collect()
  //*****************************调用已定义分区  确定分区号
  val par = new ZPartition(teachname)
  //（（学科，老师），出现的次数）   ********用partitionBy来调用  通过学科分区
  val res2 = reducerres.partitionBy(par)
  //再已经分好区的每个学科里面进行倒序排序求前两名
 val res3 = res2.mapPartitions(it=>{
   //迭代器需要用  ToList转换之后才能进行排序      排完序之后还要变成iterator
   it.toList.sortBy(_._2).reverse.take(TopN).toIterator   //获取前几名是灵活的   不是固定的
 })


  println(res3.collect().toBuffer)
  sc.stop()
}
//创建 自定义分区类
class ZPartition(tea:Array[String])extends Partitioner{

  //给每个老师生成一个分区号
  val map =new mutable.HashMap[String,Int]()
    var i=0  //分区号从0开始
  for(t<-tea){
    map.put(t,i)
    i+=1
  }

    //确定分区个数（分区个数就是老师的个数）
  override def numPartitions: Int = tea.length

  override def getPartition(key: Any): Int = {
    val teachername=key.asInstanceOf[(String,String)]._1   //把（学科，老师）  传进来  获取每个学科的名字
    map(teachername)   //把获取的每个学科的名称 传到上面定义的hashmap可变集合中  返回一个分区数（i）
  }
}

按老师名字分区的结果

最后在分区中排完序的结果