用Spark实现的词频统计

￡神魔★判官ぃ 2024-03-25 11:04 110阅读 0赞

#### WordCount ####

*  前言
 *  准备两个文本文件
 *  方法一
 *  方法二 优化

--------------------

## 前言 ##

`为了能直观地感受 Spark 框架的效果，接下来我们实现一个大数据学科中最常见的教学 案例 WordCount`

## 准备两个文本文件 ##

![在这里插入图片描述][167cd589ef514d149ad8f6a0251afda3.png]  
![在这里插入图片描述][2a798a415e7a406f859ce34626f2a484.png]

## 方法一 ##

package com.argonaut.bigdata.spark.core.wc
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{
        SparkConf, SparkContext}
    
    object Spark02_WordCount {
        
      def main(args: Array[String]): Unit = {
        
        //Application
        //Spark框架(一套运行应用程序的环境)
        //TODO 建立和Spark框架的连接
        //JDBC : Connection
        val sparConf = new SparkConf().setMaster("local").setAppName("WordCount") //Spark基础配置对象
        val sc = new SparkContext(sparConf)
    
        //TODO 执行业务操作
        //1.读取文件,获取一行一行的数据
        //这里解释一下，textFile返回的RDD[String]类型，
        // Scala中可以省略返回类型，编译器会自己去找
        // hello world
        val lines: RDD[String] = sc.textFile(path = "datas")
    
        //2.将一行数据进行拆分，分词
        // 扁平隐射把一行字符串变成一个一个单词
        //_.split() 等价于 x => x.split()
        // 等价于 (x) => {x.split}， 是一个匿名函数
        val words = lines.flatMap(x => x.split(" "))
    
        //3.将数据根据单词进行分组，便于统计
        val wordGroup: RDD[(String, Iterable[String])] = words.groupBy(word => word)
        //4.对分组后数据进行转换map()结构转换一般用
        val wordToCount = wordGroup.map {
        
          case (word, list) => {
        
            (word, list.size)
          }
        }
        //5.将转换结果采集到控制台打印
        val array: Array[(String, Int)] = wordToCount.collect()
        array.foreach(println)
        //TODO 关闭连接SparkContext
        sc.stop()
      }
    }

## 方法二 优化 ##

package com.argonaut.bigdata.spark.core.wc
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{
        SparkConf, SparkContext}
    
    object Spark01_WordCount {
        
      def main(args: Array[String]): Unit = {
        
        //建立spark框架连接
        //基本配置
        val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
        //sparkConf基础配置对象 sparkConf环境 local本地环境 应用名称：WordCount
        val sc = new SparkContext(sparkConf) //将sparkConf传入核心对象中
    
        //读取数据
        val lin: RDD[String] = sc.textFile("datas")
    
        //拆分数据（扁平化）
        val words: RDD[String] = lin.flatMap(_.split(" "))
    
        val wordOne = words.map(
          word => (word,1)//表示单词出现过一次
        )
    
        //Spark 提供了更方便的功能 将分组和聚合使用一个方法实现
        //reduceByKey 相同的key 可以对value数据进行reduce聚合
        val wCount = wordOne.reduceByKey(_+_) //(x,y) => {x+y}
    
        //数据采集输出
        val arr: Array[(String, Int)] = wCount.collect()
        arr.foreach(println)
    
        //关闭spark框架连接
        sc.stop()
      }
    }

![在这里插入图片描述][c731ffe06f2b4874b799f5d1022f22e5.png]

[167cd589ef514d149ad8f6a0251afda3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/f7f052ee902b44278c36061cfd2bee96.png
[2a798a415e7a406f859ce34626f2a484.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/299dc1d7a9f641639a5a0c79e4758658.png
[c731ffe06f2b4874b799f5d1022f22e5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/c6accc2517264039987d818a365a57cb.png