scala中groupBy用在partition前面还是后面

Myth丶恋晨 2021-11-04 14:40 110阅读 0赞

** 要求：根据id分组，并对utc进行排序**

val conf = new SparkConf()
          .setAppName("flow")
          .setMaster("local[*]")
          .registerKryoClasses(Array[Class[_]](A.getClass, Trip.getClass, Line.getClass, Log.getClass, LogMinor.getClass, LogData.getClass, UnConformData.getClass, LineX.getClass, MatchDataMajor.getClass))
    
        val sparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()
        //正确的写法  
        val list: List[A] = List(A(1, 234), A(1, 123), A(1, 345), A(1, 456))
        val data = sparkSession.sparkContext.parallelize(list.groupBy(_.uuid).toList) //在分片前面groupBy
        data.foreachPartition {
          partition =>
            partition.foreach(_._2.sortBy(_.utc).foreach(println))
        }
        /* 结果：
         A(1,123)
          A(1,234)
          A(1,345)
          A(1,456)*/

val conf = new SparkConf()
          .setAppName("flow")
          .setMaster("local[*]")
          .registerKryoClasses(Array[Class[_]](A.getClass, Trip.getClass, Line.getClass, Log.getClass, LogMinor.getClass, LogData.getClass, UnConformData.getClass, LineX.getClass, MatchDataMajor.getClass))
    
        val sparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()
        //错误的写法
        val list: List[A] = List(A(1, 234), A(1, 123), A(1, 345), A(1, 456))
        val data = sparkSession.sparkContext.parallelize(list)
    
        val unit: Unit = data.foreachPartition {
          var num = 0
          partition =>  //在分片里面进行的分区
            partition.toList.groupBy(_.uuid).map(_._2.sortBy(_.utc)).foreach(println)  
        }
        /**
          * 结果:
          * List(A(1,123))
          * List(A(1,234))
          * List(A(1,345))
          * List(A(1,456))
          */