函数签名:
def groupByk(implicitkt:ClassTag[K]):RDD[{K,Iterable[T]}]

groupBy算子会对RDD里面的每个元素进行遍历,给每个元素取出来作为参数传递给T, 然后根据分组规则去匹配.

groupBy会存在shuffle过程

shuffle：将不同的分区数据进行打乱重组的过程
shuffle一定会落盘。可以在local模式下执行程序，通过4040看效果。
作用:
按照func的返回值进行分组.
func返回值作为 key, 对应的值放入一个迭代器中. 返回的 RDD: RDD[(K, Iterable[T])
每组内元素的顺序不能保证, 并且甚至每次调用得到的顺序也有可能不同.

groupBy算子返回值是元祖,key是当前组,value是当前组的所有元素.

image.png

例子

根据偶数和奇数分组

  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.{SparkConf, SparkContext}
  5. /**
  6.  *
  7.  * Desc: 转换算子-groupBy
  8.  * -按照指定的规则，对RDD中的元素进行分组
  9.  */
  10. object Spark05_Transformation_groupBy {
  11.   def main(args: Array[String]): Unit = {
  12.     //创建SparkConf并设置App名称
  13.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  15.     //创建SparkContext，该对象是提交Spark App的入口
  16.     val sc: SparkContext = new SparkContext(conf)
  18.     val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
  19.     println("==============groupBy分组前================")
  20.     rdd.mapPartitionsWithIndex(
  21.       (index, datas) => {
  22.         println(index + "---->" + datas.mkString(","))
  23.         datas
  24.       }
  25.     ).collect()
  26.     /*输出:  0---->1,2,3,4
  27.             1---->5,6,7,8,9*/
  29.     //按奇数和偶数分组
  30.     val newRDD: RDD[(Int, Iterable[Int])] =
  31.       rdd.groupBy(x => x % 2) // 可以简写成     rdd.groupBy(_ % 2)
  33.     println("==============groupBy分组后================")
  34.     newRDD.mapPartitionsWithIndex(
  35.       (index, datas) => {
  36.         println(index + "---->" + datas.mkString(","))
  37.         datas
  38.       }
  39.     ).collect()
  40.     //这里有三个分区,但是是根据奇数偶数分组的,数据就只有两条,要不就是奇数要不就是偶数
  41.     //所以2号区域就是空的了.
  42.     /*
  43.     2---->
  44.     1---->(1,CompactBuffer(1, 3, 5, 7, 9))
  45.     0---->(0,CompactBuffer(2, 4, 6, 8))
  46.     */
  49.     // 关闭连接
  50.     sc.stop()
  51.   }
  52. }

按相同元素进行分组

  2. object Spark05_Transformation_groupBy222 {
  3.   def main(args: Array[String]): Unit = {
  4.     //创建SparkConf并设置App名称
  5.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  7.     //创建SparkContext，该对象是提交Spark App的入口
  8.     val sc: SparkContext = new SparkContext(conf)
  9.     val original = sc.makeRDD(List("aaa", "bbb", "ccc", "aaa", "ccc"))
  11.     val value = original.groupBy(datas => datas)
  12.     val tuples:Array[(String, Iterable[String])] = value.collect()
  14.     for (i <- tuples) {
  15.       println(i)
  16.     }
  17.     /*
  18.     (ccc,CompactBuffer(ccc, ccc))
  19.     (bbb,CompactBuffer(bbb))
  20.     (aaa,CompactBuffer(aaa, aaa))
  21.     */
  23.   }
  24. }

按把奇数和偶数值分组,并且累加

  2. object Spark05_Transformation_groupBy222 {
  3.   def main(args: Array[String]): Unit = {
  4.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  6.     val sc: SparkContext = new SparkContext(conf)
  7.     val original = sc.makeRDD(List(1, 2, 3, 4, 5, 6))
  8.     //按奇数偶数分组
  9.     val gbrdd = original.groupBy(datas => datas % 2)
  11.     val tuples: Array[(Int, Iterable[Int])] = gbrdd.collect()
  12.     val value = gbrdd.map({
  13.       //每个组的值累加
  14.       case (k, it) => (k, it.sum)
  15.     })
  16.     println(value.collect().mkString(",")) //输出:   (0,12),(1,9)
  18.   }
  19. }