Scala编写Spark分组后排序取TOP n的代码

╰半橙微兮° 2022-05-12 03:16 157阅读 0赞

1、首先来练习最简单的一种情况，数据如下所示，我们要先进行分组然后取每组中的最大值

1 89
    2 100
    2 92
    1 80
    1 90

具体代码：

//两列数据
      def top_1(): Unit = {
        val sc = TransformationOperation.getSc("groupByKey2Top")
        val lines = sc.textFile("D:\\data\\testdatas\\spark\\group2top.txt")
        lines
          .map(line => (line.split(" ")(0).toInt, line.split(" ")(1).toInt))
          //按第一列进行分组
          .groupByKey()
          .foreach(line => {
            //将每组的value转化为list进行排序
            val arrs = line._2
            val list = arrs.toList.sorted.reverse
            println(line._1 + "组最大值：" + list(0))
          })
      }

输出：

1组最大值：90  
2组最大值：100

\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*

2、第二种情况的数据如下：第一列：科目，第二列：名字，第三列：分值。

需求：求出每科的最高分(包含名字)

A zs 90
    B ww 100
    A ls 100
    C hr 80
    C js 100
    B wk 60

具体代码：

// three columns
      def top_2(): Unit = {
        val sc = TransformationOperation.getSc("groupByKey2Top")
        val lines = sc.textFile("D:\\data\\testdatas\\spark\\group2top2.txt")
        val categoryRdd = lines.map(line => {
          val category = line.split(" ")(0)
          val name = line.split(" ")(1)
          val score = line.split(" ")(2).toInt
          new Tuple2[String, Tuple2[String, Integer]](category, new Tuple2[String, Integer](name, score))
        }).groupByKey()
        /** categoryRdd
          * (B,CompactBuffer((ww,100), (wk,60)))
          * (A,CompactBuffer((zs,90), (ls,100)))
          * (C,CompactBuffer((hr,80), (js,100)))
          */
        val result = categoryRdd.map(x => {
          val category = x._1
          val scoreInfos = x._2
          var lists = List[(String, Integer)]()
          for (i <- scoreInfos) {
            val name = i._1
            val score = i._2
            lists = ((name, score)) :: lists
          }
          lists = lists.sortBy(_._2).reverse.take(1)
          new Tuple2[String, Iterable[(String, Integer)]](category, lists)
        })
        result.foreach(println(_))
      }

输出结果：

(B,List((ww,100)))  
(A,List((ls,100)))  
(C,List((js,100)))

\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*手动分割线\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*

实现的方式很多，在第二种情况我尝试过 sc.parallelize(lists).... 来代替lists = lists.sortBy(\_.\_2).reverse.take(1)将集合zhuan转化为RDD进行操作，不过因为lists的原因报错了，暂时没找到合适的解决办法，有兴趣的也可以试试然后来帮我解答解答疑惑

--------------------

欢迎加群大家一起学习大数据~群里还有很多资料可供参考~

大数据大佬菜鸡交流群 575419003