Spark大数据处理讲课笔记---Spark RDD典型案例

news/2024/2/28 1:57:52

零、本节学习目标

  1. 利用RDD计算总分与平均分
  2. 利用RDD统计每日新增用户
  3. 利用RDD实现分组排行榜

一、利用RDD计算总分与平均分

(一)提出任务

  • 针对成绩表,计算每个学生总分和平均分

 

(二)实现思路

  • 读取成绩文件,生成lines;定义二元组成绩列表;遍历lines,填充二元组成绩列表;基于二元组成绩列表创建RDD;对rdd按键归约得到rdd1,计算总分;将rdd1映射成rdd2,计算总分与平均分。

(三)准备工作

1、启动HDFS服务

  • 执行命令:start-dfs.sh

2、启动Spark服务

  • 执行命令:start-all.sh

3、在本地创建成绩文件

  • /home里创建scores.txt文件

4、将成绩文件上传到HDFS

  • 在HDFS上创建/scoresumavg/input目录,将成绩文件上传至该目录

 

四)完成任务

1、在Spark Shell里完成任务

(1)读取成绩文件,生成RDD

  • 执行命令:val lines = sc.textFile("hdfs://master:9000/scoresumavg/input/scores.txt")

 

(2)定义二元组成绩列表

  • 执行命令:import scala.collection.mutable.ListBuffer
  • 执行命令:val scores = new ListBuffer[(String, Int)]()

 

(3)利用RDD填充二元组成绩列表

lines.collect.foreach(line => {                 val fields = line.split(" ")                  scores.append((fields(0), fields(1).toInt))   scores.append((fields(0), fields(2).toInt))   scores.append((fields(0), fields(3).toInt))   
})
scores.foreach(println)                                        
  • 执行上述代码

 

(4)基于二元组成绩列表创建RDD

  • 执行命令:val rdd = sc.makeRDD(scores);

 

 

(5)对rdd按键归约得到rdd1,计算总分

  • 执行命令:val rdd1 = rdd.reduceByKey(_ + _)

(6)将rdd1映射成rdd2,计算总分与平均分

  • 执行命令:val rdd2 = rdd1.map(score => (score._1, score._2, (score._2 / 3.0).formatted("%.2f")))

 

2、在IntelliJ IDEA里完成任务

(1)打开RDD项目

  • SparkRDDDemo

(2)创建计算总分平均分对象

  • net.cl.rdd包里创建day07子包,然后在子包里创建CalculateSumAvg对象
package net.huawei.rdd.day07import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable.ListBuffer/*** 功能:统计总分与平均分* 作者:华卫* 日期:2023年05月11日*/
object CalculateSumAvg {def main(args: Array[String]): Unit = {// 创建Spark配置对象val conf = new SparkConf().setAppName("CalculateSumAvg ") // 设置应用名称.setMaster("local[*]") // 设置主节点位置(本地调试)// 基于Spark配置对象创建Spark容器val sc = new SparkContext(conf)// 读取成绩文件,生成RDDval lines = sc.textFile("hdfs://master:9000/scoresumavg/input/scores.txt")// 定义二元组成绩列表val scores = new ListBuffer[(String, Int)]()// 利用RDD填充二元组成绩列表lines.collect.foreach(line => {val fields = line.split(" ")scores.append((fields(0), fields(1).toInt))scores.append((fields(0), fields(2).toInt))scores.append((fields(0), fields(3).toInt))})// 基于二元组成绩列表创建RDDval rdd = sc.makeRDD(scores);// 对rdd按键归约得到rdd1,计算总分val rdd1 = rdd.reduceByKey(_ + _)// 将rdd1映射成rdd2,计算总分与平均分val rdd2 = rdd1.map(score => (score._1, score._2, (score._2 / 3.0).formatted("%.2f")))// 在控制台输出rdd2的内容rdd2.collect.foreach(println)// 将rdd2内容保存到HDFS指定位置rdd2.saveAsTextFile("hdfs://master:9000/scoresumavg/output")// 关闭Spark容器sc.stop()    }
}

(3)运行程序,查看结果

  • 运行程序CalculateSumAvg,控制台结果

  • 查看HDFS的结果文件

 

二、利用RDD统计每日新增用户

(一)提出任务

  • 已知有以下用户访问历史数据,第一列为用户访问网站的日期,第二列为用户名。
2023-05-01,mike
2023-05-01,alice
2023-05-01,brown
2023-05-02,mike
2023-05-02,alice
2023-05-02,green
2023-05-03,alice
2023-05-03,smith
2023-05-03,brian

 

 

  • 现需要根据上述数据统计每日新增的用户数量,期望统计结果。
2023-05-01新增用户数:3
2023-05-02新增用户数:1
2023-05-03新增用户数:2
  • 即2023-05-01新增了3个用户(分别为mike、alice、brown),2023-05-02新增了1个用户(green),2023-05-03新增了两个用户(分别为smith、brian)。

(二)实现思路

  • 使用倒排索引法,若将用户名看作关键词,访问日期看作文档ID,则用户名与访问日期的映射关系如下图所示。

  • 若同一个用户对应多个访问日期,则最小的日期为该用户的注册日期,即新增日期,其他日期为重复访问日期,不应统计在内。因此每个用户应该只计算用户访问的最小日期即可。如下图所示,将每个用户访问的最小日期都移到第一列,第一列为有效数据,只统计第一列中每个日期的出现次数,即为对应日期的新增用户数。

 

(三)准备工作

1、在本地创建用户文件

  • /home目录里创建users.txt文件

2、将用户文件上传到HDFS指定位置

  • 先创建/newusers/input目录,再将用户文件上传到该目录

(四)完成任务

1、在Spark Shell里完成任务

(1)读取文件,得到RDD

  • 执行命令:val rdd1 = sc.textFile("hdfs://master:9000/newusers/input/users.txt")

 

(2)倒排,互换RDD中元组的元素顺序

val rdd2 = rdd1.map(line => {val fields = line.split(",")(fields(1), fields(0))}
)
rdd2.collect.foreach(println)
  • 执行上述语句

 

(3)倒排后的RDD按键分组

  • 执行命令:val rdd3 = rdd2.groupByKey()

 

(4)取分组后的日期集合最小值,计数为1

  • 执行命令:val rdd4 = rdd3.map(line => (line._2.min, 1))

(5)按键计数,得到每日新增用户数

  • 执行命令:val result = rdd4.countByKey()

  • 执行命令:result.keys.foreach(key => println(key + "新增用户:" + result(key)))

(6)让输出结果按日期升序

  • 映射不能直接排序,只能让键集转成列表之后先排序,再遍历键集输出映射
  • 执行命令:val keys = result.keys.toList.sorted,让键集升序排列

 

2、在IntelliJ IDEA里完成任务

(1)打开RDD项目

  • SparkRDDDemo

(2)创建统计新增用户对象

  • net.cl.day07包里创建CountNewUsers对象
package net.cl.rdd.day07import org.apache.spark.{SparkConf, SparkContext}object CountNewUsers {def main(args: Array[String]): Unit = {// 创建Spark配置对象val conf = new SparkConf().setAppName("CountNewUsers") // 设置应用名称.setMaster("local[*]") // 设置主节点位置(本地调试)// 基于Spark配置对象创建Spark容器val sc = new SparkContext(conf)// 读取文件,得到RDDval rdd1 = sc.textFile("hdfs://master:9000/newusers/input/users.txt")// 倒排,互换RDD中元组的元素顺序val rdd2 = rdd1.map(line => {val fields = line.split(",")(fields(1), fields(0))})// 倒排后的RDD按键分组val rdd3 = rdd2.groupByKey()// 取分组后的日期集合最小值,计数为1val rdd4 = rdd3.map(line => (line._2.min, 1))// 按键计数,得到每日新增用户数val result = rdd4.countByKey()// 让统计结果按日期升序val keys = result.keys.toList.sortedkeys.foreach(key => println(key + "新增用户:" + result(key)))// 停止Spark容器sc.stop()}
}

(3)运行程序,查看结果

  • 运行程序CountNewUsers,控制台结果

 

三、利用RDD实现分组排行榜

(一)提出任务

  • 分组求TopN是大数据领域常见的需求,主要是根据数据的某一列进行分组,然后将分组后的每一组数据按照指定的列进行排序,最后取每一组的前N行数据。
  • 有一组学生成绩数据
张三丰 90
李孟达 85
张三丰 87
王晓云 93
李孟达 65
张三丰 76
王晓云 78
李孟达 60
张三丰 94
王晓云 97
李孟达 88
张三丰 80
王晓云 88
李孟达 82
王晓云 98
  • 同一个学生有多门成绩,现需要计算每个学生分数最高的前3个成绩,期望输出结果如下所示
    张三丰:94 90 87
    李孟达:88 85 82
    王晓云:98 97 93
    

(二)实现思路

  • 使用Spark RDD的groupByKey()算子可以对(key, value)形式的RDD按照key进行分组,key相同的元素的value将聚合到一起,形成(key, value-list),将value-list中的元素降序排列取前N个即可。

(三)准备工作

1、在本地创建成绩文件

  • /home目录里创建grades.txt文件

 

2、将成绩文件上传到HDFS上指定目录

  • grades.txt上传到HDFS的/topn/input目录

 

(四)完成任务

1、在Spark Shell里完成任务

(1)读取成绩文件得到RDD

  • 执行命令:val lines = sc.textFile("hdfs://master:9000/topn/input/grades.txt")
  • grades.txt上传到HDFS的/topn/input目录

 

(2)利用映射算子生成二元组构成的RDD

val grades = lines.map(line => {val fields = line.split(" ")(fields(0), fields(1))})
grades.collect.foreach(println)
  • 执行上述代码

 

(3)按键分组得到新的二元组构成的RDD

  • 执行命令:val groupGrades = grades.groupByKey()

 

(4)按值排序,取前三

val top3 = groupGrades.map(item => {val name = item._1val top3 = item._2.toList.sortWith(_ > _).take(3)(name, top3)})
top3.collect.foreach(println)
  • 执行上述代码

 

(5)按指定格式输出结果

top3.collect.foreach(line => {val name = line._1var scores = ""line._2.foreach(score => scores = scores + " " + score)println(name + ":" + scores)
})
  • 执行上述代码

 

  • 其实,代码可以优化

 

2、在IntelliJ IDEA里完成任务

(1)打开RDD项目

  • SparkRDDDemo

(2)创建分组排行榜单例对象

  • net.cl.rdd.day07包里创建GradeTopN单例对象
package net.huawei.rdd.day07import org.apache.spark.{SparkConf, SparkContext}/*** 功能:成绩分组排行榜* 作者:华卫* 日期:2023年05月11日*/
object GradeTopN {def main(args: Array[String]): Unit = {// 创建Spark配置对象val conf = new SparkConf().setAppName("GradeTopN") // 设置应用名称.setMaster("local[*]") // 设置主节点位置(本地调试)// 基于Spark配置对象创建Spark容器val sc = new SparkContext(conf)// 实现分组排行榜val top3 = sc.textFile("hdfs://master:9000/topn/input/grades.txt").map(line => {val fields = line.split(" ")(fields(0), fields(1))}) // 将每行成绩映射成二元组(name, grade).groupByKey() // 按键分组.map(item => {val name = item._1val top3 = item._2.toList.sortWith(_ > _).take(3)(name, top3)}) // 值排序,取前三// 输出分组排行榜结果top3.collect.foreach(line => {val name = line._1      val scores = line._2.mkString(" ")println(name + ": " + scores)})// 停止Spark容器,结束任务sc.stop()}
}

(3)运行程序,查看结果

  • 在控制台查看输出结果

 


http://www.ppmy.cn/news/75990.html

相关文章

java实现url链接的补全,获取到的链接是以/或 ./ 开头的相对链接,不是以http开头的,需要补全

一、实现的目标 在使用爬虫获取网页html数据时,解析到的链接是/或./ 开头的相对链接,不是以http开头的链接,如:/picture/0/cca65350643c441e80d390ded3975db0.png 。此时需要完成对该链接的补全,以得到正确的链接。 二、实现思路 对比完整的url链接和相对链接,进行分析,…

自动化测试框架搭建步骤教程

说起自动化测试,我想大家都会有个疑问,要不要做自动化测试? 自动化测试给我们带来的收益是否会超出在建设时所投入的成本,这个嘛别说是我,即便是高手也很难回答,自动化测试的初衷是美好的,而测试…

WebLogic:如何查看补丁版本

可以使用 /weblogic/bea/OPatch/下的opatch命令: /opatch lsinventory 执行结果: VOlogiciEDSP-APP-D-269:/opatch lsinventory Oracle Interim Patch Installer version 13.9.4_2_8 Copyright (c) 2023, Oracle Corporation. All rights reserved. Orac…

PHP程序员在外包公司的工作内容是什么,我来跟大伙聊一聊

今天呢,我要跟大家说一下,我在上班的主要工作内容。希望能为大家提供一些参考,让大家了解在外包公司,PHP程序员主要做些什么工作。 我们还会涉及到其他项目,比如Web开发、移动应用开发、数据分析和处理等。不同的项目…

群岛大战(C++)

群岛大战 英文题目:Problem StatementConstraintsInputOutputSample 1InputOutput Sample 2InputOutput Sample 3InputOutput 中文题目:问题陈述约束输出样本1输入输出 样本2输入输出 示例3输入输出 代码 英文题目: Problem Statement Ther…

一文把 JavaScript 中的 this 聊得明明白白

文章目录 1.this 是什么?2.this的指向2.1 全局上下文的 this 指向2.2 函数(普通函数)上下文中的 this 指向2.3 事件处理程序中的 this 指向2.4 以对象的方式调用时 this 的指向2.5 构造函数中的 this 指向2.6 在 类上下文中 this 的指向。2.7…

Flink第五章:处理函数

系列文章目录 Flink第一章:环境搭建 Flink第二章:基本操作. Flink第三章:基本操作(二) Flink第四章:水位线和窗口 Flink第五章:处理函数 文章目录 系列文章目录前言一、基本处理函数(ProcessFunction)二、按键分区处理函数(KeyedProcessFunction)1.处理…

一、尚医通登录需求

文章目录 一、登录需求1、登录效果2、登录需求 二、登录1,搭建service-user模块1.1 搭建service-user模块1.2 修改配置1.3 启动类1.4 配置网关 2、添加用户基础类2.1 添加model2.2 添加Mapper2.3 添加service接口及实现类2.4 添加controller 3、登录api接口3.1 添加…

linux系统升级/更新OpenSSL版本操作流程记录

问题描述:有时 OpenSSL 版本过老升级,或者需要更新 OpenSSL 版本 1. 登录 linux 系统后输入 openssl version 查看现在使用的版本 我的输入后版本信息为:OpenSSL 1.1.1g FIPS 21 Apr 2020 ,可以看到是一年前更新版本,…

深入浅出 SQL Server CDC 数据同步

简介 SQL Server 是一款老牌关系型数据库,自 1988 年由 Microsoft、Sybase 和 Ashton-Tate 三家公司共同推出,不断迭代更新至今,拥有相当广泛的用户群体。 如今,我们提到 SQL Server 通常指 Microsoft SQL Server 2000 之后的版本。 SQL S…

网络安全里主要的岗位有哪些?小白如何快速入门学习黑客?

入门Web安全、安卓安全、二进制安全、工控安全还是智能硬件安全等等,每个不同的领域要掌握的技能也不同。 当然入门Web安全相对难度较低,也是很多人的首选。主要还是看自己的兴趣方向吧。 本文就以下几个问题来说明网络安全大致学习过程👇 网…

什么是jquery jq的基本使用

JQuery的概述 jQuery是一个快速的,简洁的javaScript库,使用户能更方便地处理HTML documents、events、实现动画效果,并且方便地为网站提供AJAX交互。 jQuery能够使用户的html页保持代码和html内容分离,也就是说&#xff0c…

Java高并发核心编程—CAS与JUC原子类

注:本笔记是阅读《Java高并发核心编程卷2》整理的笔记! CAS原理 JUC原子类一Atomic 基本原子类 数组原子类 引用原子类 字段更新原子类 AtomicInteger 线程安全原理 引用类型原子类 属性更新原子类 ABA问题 提升高并发场景下CAS提作的性能 以空间换时间:…

【饿了么UI】elementUI密码框图标实现睁眼和闭眼效果(阿里巴巴iconfront图标库vue项目本地引用)

elementUI中输入框的密码框属性, 默认是一个始终睁眼的图标,测试今天提bug要有闭眼效果(无大语)… 因为elementUI中的icon没有闭眼的,所以还要去iconfront下载引入 效果图: 点击后 一、下载图标 http…

Spring Security 中的过滤器链是什么?它的作用是什么

Spring Security是一个安全框架,它提供了强大的安全保护功能,可以帮助开发者更加方便地实现应用程序的安全性。Spring Security中的过滤器链是其中一个非常重要的部分,它起到了非常重要的作用。本文将介绍什么是Spring Security中的过滤器链&…

Linux基础内容(21)—— 进程消息队列和信号量

Linux基础内容(20)—— 共享内存_哈里沃克的博客-CSDN博客 目录 1.消息队列 1.定义 2.操作 2.信号量 1.定义 2.细节 3.延申 4.操作 3.IPC的特点共性 1.消息队列 1.定义 定义:是操作系统提供的内核级队列 2.操作 msgget:…

AC规则-4-规则和冲突解决

3.3 Introduction to Access Control Rule Conflict Resolution 3.3 访问控制规则冲突解决简介 本节从高层次讨论访问控制规则冲突解决。 本文档稍后会提供更多详细信息。 规则的优先级不是基于它在其他规则中的阅读顺序。 管理冲突规则的策略基于三个基本原则(…

【Android】(最新)跑马灯文字水平滚动(79/100)

先上效果: Android系统中TextView实现跑马灯效果,必须具备以下几个条件: android:singleLine“true”android:ellipsize“marquee”android:marqueeRepeatLimit“marquee_forever”TextView必须单行显示,即内容必须超出TextView…

【LeetCode: 10. 正则表达式匹配 | 暴力递归=>记忆化搜索=>动态规划 】

🚀 算法题 🚀 🌲 算法刷题专栏 | 面试必备算法 | 面试高频算法 🍀 🌲 越难的东西,越要努力坚持,因为它具有很高的价值,算法就是这样✨ 🌲 作者简介:硕风和炜,…

Linux——网络基础1

所谓 "局域网" 和 "广域网" 只是一个相对的概念. 比如, 我们有 "天朝特色" 的广域网, 也可以看做一个比较大的局域 网. 操作系统内部存在着多种协议,那么操作系统要管理这些协议吗? 是,管理方式:先描述,再组织。 协议本质就是软件,软件是可…
最新文章