spark3.0教程：RDD转换算子groupBy()

作者：马育民 • • 阅读：10284

说明

相当于 scala 集合中的 groupBy()

声明

按照指定的规则对集合的元素进行分组，返回 Map，key是传入函数的计算结果，value是 Iterable 类型（分到该组的元素）

指定分组规则：

def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])] = withScope {
    groupBy[K](f, defaultPartitioner(this))
}

指定分组规则，指定新分区数量，默认分区规则 HashPartitioner：

def groupBy[K](
      f: T => K,
      numPartitions: Int)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])] = withScope {
    groupBy(f, new HashPartitioner(numPartitions))
}

指定分组规则，指定分区规则：

def groupBy[K](f: T => K, p: Partitioner)(implicit kt: ClassTag[K], ord: Ordering[K] = null)
      : RDD[(K, Iterable[T])] = withScope {
    val cleanF = sc.clean(f)
    this.map(t => (cleanF(t), t)).groupByKey(p)
}

底层调用的是： groupByKey()

例子

实现 wordcount

val sparConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
val sc = new SparkContext(sparConf)

val rdd:RDD[String] = sc.textFile("data/wordcount.txt")

val rdd2: RDD[String] = rdd.flatMap(_.split(" "))
val rdd3: RDD[(String, Iterable[String])] = rdd2.groupBy(e => e)
val resRDD: RDD[String] = rdd3.map(e => s"${e._1},${e._2.size}")

resRDD.collect().foreach(println)
resRDD.saveAsTextFile("res")
sc.stop()

按照姓氏对姓名进行分组

val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)

val rdd:RDD[String] = sc.makeRDD( List("李雷","李四","韩梅梅","lili","lucy"),2 )

val resRDD = rdd.groupBy(_.take(1))

resRDD.foreach(println)
sc.stop()

执行结果：

(韩,CompactBuffer(韩梅梅))
(李,CompactBuffer(李雷, 李四))
(l,CompactBuffer(lili, lucy))

map中的 key 是 lambda表达式 _.take(1) 计算的结果
value 是分到该组中的数据

分组前后，在分区中的数据

分组前的数据，可能不在分组后的分区中

一个分区中可能有一个组的数据，也可能有多个组的数据

极限情况下，数据可能被分在同一个分区中

val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)

val rdd:RDD[String] = sc.makeRDD( List("lili","李雷","韩梅梅","李四","lucy"),2 )
rdd.saveAsTextFile("res")
val resRDD = rdd.groupBy(_.take(1))
resRDD.saveAsTextFile("res2")
sc.stop()

从执行结果可知：

分组前：

  • lili、李雷 在分区0中
  • 韩梅梅、李四、lucy 在分区1中

分组后：

  • 李雷、李四、lili, lucy 在分区0中
  • 韩梅梅 在分区1中

原因：根据姓氏分组，不同分区 但 姓氏相同的，肯定要分到一个分区中

案例：统计 IP 出现的次数

数据：

49.74.3.207 - - [14/Jun/2021:00:01:12 +0800] "GET /show_1EF4h7aLPSFB.html HTTP/1.1" 200 17300
121.226.148.220 - - [14/Jun/2021:00:01:14 +0800] "GET /show_1C7cGrpMhQ2.html HTTP/1.1" 200 21979
117.95.183.206 - - [14/Jun/2021:00:02:41 +0800] "GET /show_1EF1cWTFezUR.html HTTP/1.1" 200 15432
157.55.39.186 - - [14/Jun/2021:00:03:58 +0800] "GET /show_1EF5fipw0O0n.html HTTP/1.1" 200 20482
203.208.60.42 - - [14/Jun/2021:00:04:12 +0800] "GET /show_1IX51xrrA8z.html HTTP/1.1" 200 16986
203.208.60.29 - - [14/Jun/2021:00:04:14 +0800] "POST /ajaxApi/addViewNum HTTP/1.1" 200 40
203.208.60.114 - - [14/Jun/2021:00:04:48 +0800] "GET /show_1EF3uYu8CMSP.html HTTP/1.1" 200 20777

代码

val sparConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparConf)

val rdd:RDD[String] = sc.textFile("C:\\Users\\yumin\\Desktop\\localhost_access_log.2021-06-14.txt")
val mapRDD: RDD[String] = rdd.map(_.split(" ")(0))
val groupRDD = mapRDD.groupBy(e=>e)
//    groupRDD.map( (_._1,_._1)) //不允许这么写
val resRDD = groupRDD.map(e=>(e._1,e._2.size))
resRDD.foreach(println)
sc.stop()

原文出处：/show_1IX1KMZjwqnz.html

吉ICP备15003043号