Spark-核心编程（三）使用RDD算子完成WordCount

男娘i 2022-09-02 13:50 226阅读 0赞

#### 其他WordCount解决方法 ####

// 转换算子
    // groupBy
    def wordcount1(sc: SparkContext): Unit = { 
        sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .groupBy(word => word)
        .mapValues(iter => iter.size)
        .collect()
        .foreach(println)
    }
    
    // groupByKey
    def wordcount2(sc: SparkContext): Unit = { 
        sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(_ -> 1)
        .groupByKey()
        .mapValues(iter => iter.size)
        .collect()
        .foreach(println)
    }
    
    // reduceByKey
    def wordcount3(sc: SparkContext): Unit = { 
        sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(_ -> 1)
        .reduceByKey(_ + _)
        .collect()
        .foreach(println)
    }
    
    // aggregateByKey
    def wordcount4(sc: SparkContext): Unit = { 
        sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(_ -> 1)
        .aggregateByKey(0)(_ + _, _ + _)
        .collect()
        .foreach(println)
    }
    
    // foldByKey
    def wordcount5(sc: SparkContext): Unit = { 
        sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(_ -> 1)
        .foldByKey(0)(_ + _)
        .collect()
        .foreach(println)
    }
    
    // combineByKey
    def wordcount6(sc: SparkContext): Unit = { 
        sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(_ -> 1)
        .combineByKey(
            x => x,
            (x: Int, y) => x + y,
            (x: Int, y: Int) => x + y
        )
        .collect()
        .foreach(println)
    }
    
    // 行动算子
    // countByKey
    def wordcount7(sc: SparkContext): Unit = { 
        val rdd: collection.Map[String, Long] = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(_ -> 1)
        .countByKey()
    
        println(rdd)
    }
    
    // countByValue
    def wordcount8(sc: SparkContext): Unit = { 
        val rdd: collection.Map[String, Long] = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .countByValue()
    
        println(rdd)
    }
    
    // reduce
    def wordcount9(sc: SparkContext): Unit = { 
        val rdd: mutable.Map[String, Long] = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(word => { 
            mutable.Map[String, Long]((word, 1))
        })
        .reduce((map1, map2) => { 
            map2.foreach { 
                case (word, count) => { 
                    val newCount = map1.getOrElse(word, 0L) + count
                    map1.update(word, newCount)
                }
            }
            map1
        })
    
        println(rdd)
    }
    
    // aggregate
    def wordcount10(sc: SparkContext): Unit = { 
        val rdd: mutable.Map[String, Long] = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(word => { 
            Map((word, 1))
        })
        .aggregate(mutable.Map[String, Long]())(
            (map1, map2) => { 
                map2.foreach { 
                    case (word, count) => { 
                        val newCount = map1.getOrElse(word, 0L) + count
                        map1.update(word, newCount)
                    }
                }
                map1
            },
            (map1, map2) => { 
                map2.foreach { 
                    case (word, count) => { 
                        val newCount = map1.getOrElse(word, 0L) + count
                        map1.update(word, newCount)
                    }
                }
                map1
            }
        )
    
        println(rdd)
    }
    
    
    // fold
    def wordcount11(sc: SparkContext): Unit = { 
        val rdd: mutable.Map[String, Long] = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        .flatMap(_.split(" "))
        .map(word => { 
            mutable.Map((word, 1L))
        })
        .fold(mutable.Map[String, Long]())(
            (map1, map2) => { 
                map2.foreach { 
                    case (word, count) => { 
                        val newCount: Long = map1.getOrElse(word, 0L) + count
                        map1.update(word, newCount)
                    }
                }
                map1
            }
        )
    
        println(rdd)
    }