8ed48d4242cf222e6b81a979648b05b9.png

注:图片如果损坏,点击文章链接:https://www.toutiao.com/i6814778610788860424/

编写类似MapReduce的案例-单词统计WordCount

要统计的文件为Spark的README.md文件

f358c354ef3264e269da016e1d27881a.png

分析逻辑:

1. 读取文件,单词之间用空格分割

2. 将文件里单词分成一个一个单词

3. 一个单词,计数为1,采用二元组计数word ->(word,1)

4. 聚合统计每个单词出现的次数

RDD的操作

8af7b59b48657602c7ac3904d29f66fe.png

1.读取文件:

sc.textFile("file:///opt/modules/spark/README.md")

注意:textFile里面的路径,如果没有指定schema,那么默认的话是从HDFS文件系统读取数据,如果不加file://就是/opt/modules/spark/README.md的路径,是从HDFS对应目录下读取

接收变量是res0

320f9356d3aa8ab0a41a1db27ef93619.png

res0方法查看

2ae8ca3f9208f340b1951b60c6cda13b.png

可以用一个变量接收

799b084d6e2473f23d551b8e11dc08d8.png

rdd.count-->统计RDD里有多少条数据

b813dcb7a96b03057c2349472c8d250a.png

rdd.first--> 取RDD的第一条数据

7d6b55f003ec9b1ecd45ed8a42fe3896.png

可以对比源文件看到第一条数据

0f19e9c0c16471306af1fc28b3f230d0.png

rdd.take(10)--> 取RDD的前10条数据,也可以对比源文件查看

8137658a5d6a58539dbf9cc9e59fb4a6.png

2. 将文件中的数据分成一个一个的单词

59e55457c6a751e658431394657785c0.png

map和flatMap返回类型不一致,返回结果类型是RDD[String]和RDD[Array[String]]

bc615e228e061db2cb1f8c47a7e0b566.png

可以使用collect方法,查看结果

变量名

9a15a56ba0ac69e922c8219ee8d9e542.png

collect方法

29bf0aca81a77c5205a65990aba225b9.png

map和flatMap返回结果的不同之处:flatMap会进行扁平化操作

mapRDD = rdd.map(line => line.split(" "))

第一个元素:Array("#","Apache","spark")

56fd460a6aeca3192c3c5d356e96bbd5.png

第三个元素:Array("Spark","is")

55ea2671b6581d60e6f30985c3ea7113.png

flatmapRDD = rdd.flatMap(line => line.split(" "))

第一个元素:"#"

15d09cfa87f3deebd270c00bc3a72b59.png

所以我们选择flatMap,而不是map

val flatMapRDD = rdd.flatMap(line => line.split(" "))

//flatMapRDD: org.apache.spark.rdd.RDD[String]

去除空的字符串的操作

flatMapRDD.filter(word => word.nonEmpty)

0445630aa08cf582e46d1fa9e34ed888.png

3. 将每个单词进行计数

val mapRDD = flatMapRDD.map(word => (word,1))

7b151249388fa966905ed5bdaa033dc4.png

返回类型//mapRDD: org.apache.spark.rdd.RDD[(String, Int)]

d4c2c179389ad9f2a855584a26006414.png

4.将相同的单词放在一起进行value值得聚合

val reduceRDD = mapRDD.reduceByKey((a,b) => a + b)

//reduceRDD: org.apache.spark.rdd.RDD[(String, Int)]

e0ab9a5ffd421d55aeb974c711b3b082.png

查看对比下(reduceByKey前后两个变量的collect)

f7bde2f8737e24c5626ae75a5ee36ae2.png

链式编程写法:

val result = sc.textFile("file:///opt/modules/o2o23/spark/README.md").flatMap(line => line.split(" ")).filter(word => word.nonEmpty).map(word => (word,1)).reduceByKey((a,b) => a + b).collect

1e0efca564c410f9d70c17f54f12b5d8.png

链式编程简化写法:

val result1 = sc.textFile("file:///opt/modules/o2o23/spark/README.md").flatMap(_.split(" ")).filter(_.nonEmpty).map((_,1)).reduceByKey(_+_).collect

0613940bafb63702e38d6072ff02f0b2.png
Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐