用Spark实现的词频统计_使用spark完成词频分析

前言

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

准备两个文本文件

方法一

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()
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

方法二 优化

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()
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36