LoveCount统计“爱”——Scala-Spark实战应用案例

前言：

首先，在这里祝大家七夕快乐！！！与良人、佳人共度七夕浪漫时光！！！
单身Dogs也要快乐，用你们的五指菇凉把代码敲起来，博文写起来！
（你敲代码的样子像极了爱情！！！）
话不多说，开始我们的爱情故事！
七月七日，晴。这是我与Scala、Spark相识的第11天，虽然只有短短的11天，但是我们彼此都真切且深刻地了解了对方——她知道我的长短，就像我知道她的深浅！今天是我们的第一个七夕，所以我要以特殊的方式——一篇博文记录下我们的爱情！当然，以后还会有N多个七夕一起度过，我知道，我们将会相伴一生！

需求：

1、统计txt文件中love单词出现的次数；

2、txt文件中哪个单词出现的次数最多？

代码实现：

package cases

import org.apache.spark.{SparkConf, SparkContext}

object LoveCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("LoveCount").setMaster("local")
    val sc = new SparkContext(conf)
    val wordRDD = sc.textFile("dir/love.txt").flatMap(line => {
      val words: Array[String] = line.split(" ")
      words
    })
    val l1: Long = wordRDD.filter(word => word.equals("Love")).count()
    println(l1)
    val l2: Array[(String, Int)] = wordRDD.map(word => (word, 1)).reduceByKey(_ + _).sortBy(item => item._2, false).take(1)
    println(l2.toBuffer)
      sc.stop()
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

结果输出：

love出现了520次；文件中出现最多的单词是love
像极了爱情！！！

案例代码解析：

val conf = new SparkConf().setAppName("LoveCount").setMaster("local")
1

创建SparkConf对象（模块代码），相当于MR中配置必须传入参数；

setAppName —> 设置任务的名称，不设置默认是一个UUID所产生的名字；

setMaster —> 设置运行模式

不提交到集群中运行，即打jar包提交集群的操作，需要设置一个方法setMaster。这个方法主要是设置本地执行即本地模式，提供参数赋值：

• “local” —> 本地运行提供一个线进行任务处理

• “local[数值]” —> 开启相应数值的线程模拟spark集群运行任务

• “local[*]” —> 开启相应数值的线程模拟spark集群运行任务

  注意：“local[数值]”与“local[*]”两者的区别在于：前者[数值]根据数值决定个数，而[*]是当前空闲多少线程就用多少线程执行

//创建一个sparkContext对象，通过SparkContext对象进行数据处理操作，程序的入口点
val sc = new SparkContext(conf)
1
2

val wordRDD: RDD[String] = sc.textFile("dir/love.txt").flatMap(line => {
      val words: Array[String] = line.split(" ")
      words
    })
//以上为Scala简写，拆分如下：
1
2
3
4
5

textFile读取文件，参数是String类型，传入对应路径即可：textFile("文件路径")

 val lines: RDD[String] = sc.textFile(args(0))
1

切分读取到的数据

 val words: RDD[String] = lines.flatMap(_.split(" "))
1

flatMap遍历扁平化处理（这个flatMap和Scala中flatMap是一样的），flatMap方法可以对集合中数据进行扁平化处理，将集合内容的数据取出来返回一个全新的RDD

//Love出现的次数
val l1: Long = wordRDD.filter(word => word.equals("Love")).count()
1
2

filter对l1中每一个元素进行过滤操作（对每一个元素进行Boolean类型表达式的验证），如果结果为true就保留数据，如果结果为false则丢弃数据；

count()：统计单词出现的次数

//txt文件中哪个单词出现的次数最多？
val l2: Array[(String, Int)] = wordRDD.map(word => (word, 1)).reduceByKey(_ + _).sortBy(item => item._2, false).take(1)
1
2

map对wordRDD中每一个元素进行遍历并加以计算，返回一个新的RDD；

reduceByKey：相同key为一组进行求和计算，计算value值

sortBy：对结果进行排序，和Scala中sortBy不一样，Spark中sortBy算子多了一个参数可以设置升降序，默认升序，false就是降序。

take(1)：返回一个由数据集的前1个元素组成的数组

sc.stop()//停止SparkContext，结束任务回收资源
1

案例中所用的RDD算子

flatMap(func)

类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）

filter(func)

返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成

count()

返回RDD的元素个数

map(func)

返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成

reduceByKey(func,[numTasks])

在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置

sortBy(func,[ascending], [numTasks])

在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD。在排序前可以将数据通过f: (T) => K 函数进行处理之后按照f: (T) => K 函数处理之后结果进行排序，默认是升序，false降序

take(n)

返回一个由数据集的前n个元素组成的数组

总结

人生若只如初见！知性、曼妙，似火、似冰、似久别故人、似擦肩过客！像极了爱情！！！