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

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

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

分析逻辑:

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

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

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

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

RDD的操作

1.读取文件:

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

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

接收变量是res0

res0方法查看

可以用一个变量接收

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

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

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

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

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

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

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

变量名

collect方法

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

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

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

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

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

第一个元素:"#"

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

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

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

去除空的字符串的操作

flatMapRDD.filter(word => word.nonEmpty)

3. 将每个单词进行计数

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

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

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

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

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

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

链式编程写法:

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

链式编程简化写法:

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

 

最新文章

  1. 程序猿尤其是.NET程序员所需要注意的网站资源
  2. sql查询当前月内的所有日期
  3. 4月数据库流行度排行榜 MySQL能否追上Oracle
  4. 数据结构(KD树):HDU 4347 The Closest M Points
  5. BI商业智能项目中的若干风险要素
  6. 【cocos2d-x制作别踩白块儿】第一期:游戏介绍
  7. 剑指Offer-删除链表中重复的结点
  8. 使用ssh keys实现免验证登陆远程服务
  9. 【iCore4 双核心板_ARM】例程三十二:UART_IAP_ARM实验——更新升级STM32
  10. Python在金融量开源项目列表
  11. Linux文件系统管理
  12. elk的安装部署
  13. Git 基础 —— 安装 配置 别名 对象
  14. 七、Kafka 用户日志上报实时统计之编码实践
  15. POJ 1038 Bug Integrated Inc(状态压缩DP)
  16. JAVA多线程基础学习三:volatile关键字
  17. phpcms v9表单实现问答咨询功能
  18. cookie、localStorage和sessionStorage区别
  19. Android stateMachine分析
  20. GridView实现编辑删除

热门文章

  1. 利用Windbg分析Magicodes.IE一次错误编写导致内存剧增
  2. C++11 标准库 bind 函数
  3. 6、Redis五大数据类型---列表(List)
  4. <转>网络爬虫原理
  5. zctf_2016_note3(unlink)
  6. OpenWrt之DNS设置
  7. CF637B Chat Order 题解
  8. flink使用命令开始、停止任务
  9. SpringBoot使用@Async实现异步调用
  10. 在制造业的工业2.0中应用MOM系统