RDD 专项练习
现有分数信息文件 scores.txt
班级ID 姓名 年龄 性别 科目 成绩
12 张三 25 男 chinese 50
12 张三 25 男 math 60
12 张三 25 男 english 70
12 李四 20 男 chinese 50
12 李四 20 男 math 50
12 李四 20 男 english 50
12 王芳 19 女 chinese 70
12 王芳 19 女 math 70
12 王芳 19 女 english 70
13 张大三 25 男 chinese 60
13 张大三 25 男 math 60
13 张大三 25 男 english 70
13 李大四 20 男 chinese 50
13 李大四 20 男 math 60
13 李大四 20 男 english 50
13 王小芳 19 女 chinese 70
13 王小芳 19 女 math 80
13 王小芳 19 女 english 70
一、预处理
定义一个内部case类用于存储分数数据
创建SparkContext对象
读取数据文件，跳过第一行（标题行），并映射为Score对象(附加年龄类型)
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object SparkRDD {
// 定义一个内部case类用于存储分数数据
private case class Score(classId: Int, name: String, age: Int, gender: String, subject: String, score: Int, _type: String)
def main(args: Array[String]): Unit = {
// 创建Spark配置对象
val conf = new SparkConf()
.setAppName("spark_rdd") // 设置应用名称
.setMaster("local[4]") // 设置运行模式为本地模式，并分配4个核心
// 获取或创建SparkContext对象
val sc = SparkContext.getOrCreate(conf)
// 指定数据文件路径
val path = "file:///D:\\myOwnProject\\spark_first\\data\\scores.txt"
// 读取数据文件，跳过第一行（标题行），并映射为Score对象
val scores: RDD[Score] = sc.textFile(path, 4) // 读取文件，分区数为4
.mapPartitionsWithIndex { // 对每个分区应用索引和迭代器
case (index, iterator) => if (index == 0) iterator.drop(1) else iterator // 如果是第一个分区，则跳过第一行
}
.mapPartitions( // 对每个分区应用映射操作
_.map(line => { // 对分区中的每一行进行处理
val a = line.split("\\s+") // 按空白字符分割每行
val age = a(2).toInt // 将年龄字段转换为整数
val _type = age match { // 根据年龄设置类型
case age if age > 20 => "GT20" // 年龄大于20
case age if age == 20 => "EQ20" // 年龄等于20
case age if age < 20 => "LT20" // 年龄小于20
}
// 构造Score对象
Score(a(0).toInt, a(1), a(2).toInt, a(3), a(4), a(5).toInt, _type)
})
).cache() // 将RDD缓存到内存中，优化：RDD结果被不断使用
// 打印所有处理后的Score对象
scores.foreach(println)
//程序结束时停止SparkContext
sc.stop()
}
}
二、处理需求
1、一共有多少人参加考试？
val count1 = scores.mapPartitions(_.map(_.name)) // scores如上预处理，下同
.distinct()
.count()
println(s"${count1} 人参加考试") //6人参加考试
2、一共有多少个大于、小于、等于20岁的人参加考试？
val map = Map(("GT20", "20岁以上"), ("EQ20", "20岁"), ("LT20", "20岁以下"))
scores.mapPartitions(_.map(s => ((s.name, s._type), 1)))
.groupByKey()
.mapPartitions(_.map(s => (s._1._2, 1)))
.reduceByKey(_ + _)
.foreach(s => println(s"${map.get(s._1).get}的人数为${s._2}"))
/*
20岁以上的人数为2
20岁以下的人数为2
20岁的人数为2
*/
3、分别有多个男生和女生参加考试？
scores.mapPartitions(_.map(s => ((s.name, s.gender), 1)))
.groupByKey()
.mapPartitions(_.map(s => (s._1._2, 1)))
.reduceByKey(_ + _)
.foreach(s => println(s"${s._1}生参加考试的人数为${s._2}"))
/*
男生参加考试的人数为4
女生参加考试的人数为2
*/
4、各个班有多少人参加考试？
注意同班，同名去重
方法一：groupByKey 去重
scores.mapPartitions(_.map(s => ((s.name, s.classId), 1)))
.groupByKey()
.mapPartitions(_.map(s => (s._1._2, 1)))
.reduceByKey(_ + _)
.foreach(s => println(s"${s._1}班参加考试的人数为${s._2}"))
/*
12班参加考试的人数为3
13班参加考试的人数为3
*/
方法二：distinct去重(不推荐)
scores
.mapPartitions(
_.map(t=>(t.classId, t.name))
)
.distinct()
.map(s => (s._1, 1))
.reduceByKey(_+_)
.foreach(s => println(s"${s._1}班参加考试的人数为${s._2}"))
5、语文数学科目的平均成绩是多少？
scores
.mapPartitions(
_.collect({ // 存在非语数外科目，如何过滤
case s if s.subject.matches("chinese|math") => (s.subject, s.score)
})
).groupByKey()
.map(t => (t._1, t._2.sum * 1.0f / t._2.size))
.foreach(s => println(s"${s._1}平均分：${s._2}"))
6、单个人平均成绩是多少？
scores
.mapPartitions(_.map(t=>(t.name,t.score)))
.groupByKey()
.map(t=>(t._1,t._2.sum*1.0f/t._2.size))
.foreach(println)
7、各班平均成绩是多少？
scores
.mapPartitions(_.map(t=>(t.classId,t.score)))
.groupByKey()
.map(t=>(t._1,t._2.sum*1.0f/t._2.size))
.foreach(println)
8、各班男女生平均总成绩是多少？
scores
.mapPartitions(_.map(t=>((t.classId,t.gender),t.score)))
.groupByKey()
.map(t=>(t._1,t._2.sum*1.0f/t._2.size))
.foreach(println)
9、全校语文成绩最高分是多少？
val chineseMax = scores.mapPartitions(
_.filter(_.subject.equals("chinese"))
.map(_.score)
)
.max()
println(s"最高语文成绩为${chineseMax}")
10、各班各个科目最高和最低成绩是多少？
scores.mapPartitions(
_.map(s => ((s.subject, s.classId), (s.score, s.score)))
)
.reduceByKey(
(s1, s2) => (if (s1._1 > s2._1) s1._1 else s2._1, (if (s1._2 < s2._2) s1._2 else s2._2))
)
.foreach(s => println(s"${s._1._2}班${s._1._1}科目的最大成绩为${s._2._1},最小成绩为${s._2._2}"))
结果
13班chinese科目的最大成绩为70,最小成绩为50
12班english科目的最大成绩为70,最小成绩为50
12班chinese科目的最大成绩为70,最小成绩为50
13班english科目的最大成绩为70,最小成绩为50
12班math科目的最大成绩为70,最小成绩为50
13班math科目的最大成绩为80,最小成绩为60
11、总成绩大于 150 分的 12 班的女生有几个？
val count2 = scores.mapPartitions(
_.filter(s => s.gender == "女" && s.classId == 12)
.map(s => (s.name, s.score))
)
.reduceByKey(_ + _)
.filter(s => s._2 > 150)
.count()
println(s"总成绩大于 150 分的 12 班的女生有${count2}个")
