文章目录
- 零、本节学习目标
- 一、利用RDD计算总分与平均分
- (一)提出任务
- (二)准备工作
- 1、启动HDFS服务
- 2、启动Spark服务
- 3、在本地创建成绩文件
- 4、将成绩文件上传到HDFS
- (三)实现步骤
- 1、打开RDD项目
- 2、创建计算总分平均分对象
- 3、运行程序,查看结果
零、本节学习目标
- 利用RDD计算总分与平均分
- 利用RDD统计每日新增用户
- 利用RDD实现分组排行榜
一、利用RDD计算总分与平均分
(一)提出任务
- 针对成绩表,计算每个学生总分和平均分
姓名 | 语文 | 数学 | 英语 |
---|---|---|---|
张钦林 | 78 | 90 | 76 |
陈燕文 | 95 | 88 | 98 |
卢志刚 | 78 | 80 | 60 |
(二)准备工作
1、启动HDFS服务
- 执行命令:
start-dfs.sh
2、启动Spark服务
- 进入Spark的
sbin
目录执行命令:./start-all.sh
3、在本地创建成绩文件
- 在
/home
里创建scores.txt
文件
4、将成绩文件上传到HDFS
- 在HDFS上创建
/scoresumavg/input
目录,将成绩文件上传至该目录
(三)实现步骤
1、打开RDD项目
SparkRDDDemo
2、创建计算总分平均分对象
- 在
net.huawei.rdd
包里创建day07
子包,然后在子包里创建CalculateSumAvg
对象
package net.huawei.rdd.day07
import org.apache.spark.{SparkConf, SparkContext}
import scala.collection.mutable.ListBuffer
/**
* 功能:统计总分与平均分
* 作者:华卫
* 日期:2023年05月10日
*/
object CalculateSumAvg {
def main(args: Array[String]): Unit = {
// 创建Spark配置对象
val conf = new SparkConf()
.setAppName("CalculateSumAvg") // 设置应用名称
.setMaster("local[*]") // 设置主节点位置(本地调试)
// 基于Spark配置对象创建Spark容器
val sc = new SparkContext(conf)
// 读取成绩文件,生成RDD
val lines = sc.textFile("hdfs://master:9000/scoresumavg/input/scores.txt")
// 定义二元组成绩列表
val scores = new ListBuffer[(String, Int)]()
// 遍历lines,填充二元组成绩列表
lines.collect.foreach(line => {
val fields = line.split(" ")
scores.append(Tuple2(fields(0), fields(1).toInt))
scores.append(Tuple2(fields(0), fields(2).toInt))
scores.append(Tuple2(fields(0), fields(3).toInt))
})
// 基于二元组成绩列表创建RDD
val 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")
}
}
3、运行程序,查看结果
- 运行程序
CalculateSumAvg
,控制台结果
- 查看HDFS的结果文件