image.png


先根据flatMap进行压平,然后在根据map算子给每个词变成元祖,key就是词汇,value就是1, 然后再根据元祖的key进行分组,最后将相同key的value值进行累加.

版本1

  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.{SparkConf, SparkContext}
  6. object Spark06_WordCount1 {
  7.   def main(args: Array[String]): Unit = {
  8.     //创建SparkConf并设置App名称
  9.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  11.     //创建SparkContext，该对象是提交Spark App的入口
  12.     val sc: SparkContext = new SparkContext(conf)
  17.     //创建RDD
  18.     val rdd: RDD[String] = sc.makeRDD(List("Hello Scala", "Hello Spark", "Hello World"))
  19.    //简单版-实现方式1
  20.     //对RDD中的元素进行扁平映射
  21.     val flatMapRDD: RDD[String] = rdd.flatMap(_.split(" "))
  23.     //将映射后的数据进行结构的转换，为每个单词计数
  24.     val mapRDD: RDD[(String, Int)] = flatMapRDD.map((_,1))
  26.     //按照key对RDD中的元素进行分组   (Hello,CompactBuffer((Hello,1), (Hello,1), (Hello,1)))
  27.     val groupByRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD.groupBy(_._1)
  29.     //对分组后的元素再次进行映射  (Hello,3)
  30.     val resRDD: RDD[(String, Int)] = groupByRDD.map {
  31.       case (word, datas) => {  //模式匹配
  32.         (word, datas.size)
  33.       }
  34.     }
  38.     // 关闭连接
  39.     sc.stop()
  40.   }
  41. }

版本2

  1. import org.apache.spark.rdd.RDD
  2. import org.apache.spark.{SparkConf, SparkContext}
  4. object Spark06_WordCount2 {
  5.   def main(args: Array[String]): Unit = {
  6.     //创建SparkConf并设置App名称
  7.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  9.     //创建SparkContext，该对象是提交Spark App的入口
  10.     val sc: SparkContext = new SparkContext(conf)
  12.     //创建RDD
  13.     val rdd: RDD[String] = sc.makeRDD(List("Hello Scala", "Hello Spark", "Hello World"))
  15.     //简单版-实现方式2
  16.     //对RDD中的元素进行扁平映射
  17.     val flatMapRDD: RDD[String] = rdd.flatMap(_.split(" "))
  19.     //将RDD中的单词进行分组
  20.     val groupByRDD: RDD[(String, Iterable[String])] = flatMapRDD.groupBy(word=>word)
  22.     //对分组之后的数据再次进行映射
  23.     val resRDD: RDD[(String, Int)] = groupByRDD.map {
  24.       case (word, datas) => {
  25.         (word, datas.size)
  26.       }
  27.     }
  29.     // 关闭连接
  30.     sc.stop()
  31.   }
  32. }

版本3

  1. import org.apache.spark.rdd.RDD
  2. import org.apache.spark.{SparkConf, SparkContext}
  4. /**不建议使用
  5.  *
  6.   */
  7. object Spark06_WordCount3 {
  8.   def main(args: Array[String]): Unit = {
  9.     //创建SparkConf并设置App名称
  10.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  12.     //创建SparkContext，该对象是提交Spark App的入口
  13.     val sc: SparkContext = new SparkContext(conf)
  15.     //创建RDD
  17.     val rdd: RDD[(String, Int)] = sc.makeRDD(List(("Hello Scala", 2), ("Hello Spark", 3), ("Hello World", 2)))
  18.     //复杂版 方式1
  19.     //将原RDD中字符串以及字符串出现的次数，进行处理，形成一个新的字符串
  20.     val rdd1: RDD[String] = rdd.map {
  21.       case (str, count) => {
  22.         (str + " ") * count
  23.       }
  24.     }
  26.     //对RDD中的元素进行扁平映射
  27.     val flatMapRDD: RDD[String] = rdd1.flatMap(_.split(" "))
  29.     //将RDD中的单词进行分组
  30.     val groupByRDD: RDD[(String, Iterable[String])] = flatMapRDD.groupBy(word=>word)
  32.     //对分组之后的数据再次进行映射
  33.     val resRDD: RDD[(String, Int)] = groupByRDD.map {
  34.       case (word, datas) => {
  35.         println(word +"--"+ datas.size)
  36.         (word, datas.size)
  38.       }
  39.     }
  40.     resRDD.collect()
  41.     /*
  42. World--2
  43. Spark--3
  44. Scala--2
  45. Hello--7
  46.      */
  50.     // 关闭连接
  51.     sc.stop()
  52.   }
  53. }

版本4

  1. import org.apache.spark.rdd.RDD
  2. import org.apache.spark.{SparkConf, SparkContext}
  4. /**
  5.  * Desc: 使用groupBy完成WordCount案例
  6.  */
  7. object Spark06_WordCountOriginal {
  8.   def main(args: Array[String]): Unit = {
  9.     //创建SparkConf并设置App名称
  10.     val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
  12.     //创建SparkContext，该对象是提交Spark App的入口
  13.     val sc: SparkContext = new SparkContext(conf)
  16.     //复杂版  方式2
  17.     val rdd: RDD[(String, Int)] = sc.makeRDD(List(("Hello Scala", 2), ("Hello Spark", 3), ("Hello World", 2)))
  18.     /*先给一个元祖转成两个元祖
  19.     * ("Hello Scala", 2)==>(hello,2),(Scala,2)     ("Hello Spark", 3)==>(hello,3),(Spark,3)
  20.     */
  22.     //对RDD中的元素进行扁平映射
  23.     val flatMapRDD: RDD[(String, Int)] = rdd.flatMap {
  24.       // 模式匹配元祖, words是多个单词组成的字符串,count 是字符串出现的次数,
  25.       case (words, count) => {
  26.         words.split(" ") //对字符串进行切割,结果是字符串的数组.
  27.           .map(word => (word, count))//对每个单词转成元祖(单词,单词出现的次数)
  28.       }
  29.     }
  31.     //按照单词对RDD中的元素进行分组     (Hello,CompactBuffer((Hello,2), (Hello,3), (Hello,2)))
  32.     // _1的意思是当前元祖的第一个元素
  33.     val groupByRDD: RDD[(String, Iterable[(String, Int)])] = flatMapRDD.groupBy(_._1)
  35.     //对RDD的元素重新进行映射
  36.     val resRDD: RDD[(String, Int)] = groupByRDD.map {
  37.       case (word, datas) => {
  38.         // _2意思是:拿到元祖第二个元素进行累加.
  39.         (word, datas.map(_._2).sum)
  40.       }
  41.     }
  43.     resRDD.collect().foreach(println)
  46.     // 关闭连接
  47.     sc.stop()
  48.   }
  49. }