RDD 专项练习

图片无法加载请跳转CSDN：RDD 专项练习-CSDN博客

现有分数信息文件 scores.txt

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
班级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
需求如下：
1、一共有多少人参加考试？
2、一共有多少个大于、小于、等于20岁的人参加考试？
3、分别有多个男生和女生参加考试？
4、各个班有多少人参加考试？
5、语文和数学科目的平均成绩是多少？
6、单个人平均成绩是多少？
7、各班平均成绩是多少？
8、各班男女生平均总成绩是多少？
9、全校语文成绩最高分是多少？
10、各班各个科目最高和最低成绩是多少？
11、总成绩大于 150 分的 12 班的女生有几个？

一、预处理

定义一个内部case类用于存储分数数据
创建SparkContext对象
读取数据文件，跳过第一行（标题行），并映射为Score对象(附加年龄类型)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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、一共有多少人参加考试？

1
2
3
4
5
val count1 = scores.mapPartitions(_.map(_.name))	// scores如上预处理，下同
      .distinct()
      .count()

println(s"${count1} 人参加考试")			//6人参加考试

2、一共有多少个大于、小于、等于20岁的人参加考试？

1
2
3
4
5
6
7
8
9
10
11
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、分别有多个男生和女生参加考试？

1
2
3
4
5
6
7
8
9
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 去重

1
2
3
4
5
6
7
8
9
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去重(不推荐)

1
2
3
4
5
6
7
8
scores
    .mapPartitions(
        _.map(t=>(t.classId, t.name))
    )
    .distinct()
    .map(s => (s._1, 1))
    .reduceByKey(_+_)
    .foreach(s => println(s"${s._1}班参加考试的人数为${s._2}"))

5、语文数学科目的平均成绩是多少？

1
2
3
4
5
6
7
8
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、单个人平均成绩是多少？

1
2
3
4
5
scores
    .mapPartitions(_.map(t=>(t.name,t.score)))
    .groupByKey()
    .map(t=>(t._1,t._2.sum*1.0f/t._2.size))
    .foreach(println)

7、各班平均成绩是多少？

1
2
3
4
5
scores
    .mapPartitions(_.map(t=>(t.classId,t.score)))
    .groupByKey()
    .map(t=>(t._1,t._2.sum*1.0f/t._2.size))
    .foreach(println)

8、各班男女生平均总成绩是多少？

1
2
3
4
5
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、全校语文成绩最高分是多少？

1
2
3
4
5
6
val chineseMax = scores.mapPartitions(
    _.filter(_.subject.equals("chinese"))
    	.map(_.score)
    )
	.max()
println(s"最高语文成绩为${chineseMax}")

10、各班各个科目最高和最低成绩是多少？

1
2
3
4
5
6
7
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}"))

结果

1
2
3
4
5
6
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 班的女生有几个？

1
2
3
4
5
6
7
8
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}个")
大数据 > spark
#bigData #spark
RDD 专项练习
https://leaf-domain.gitee.io/2024/08/04/spark04/
作者
叶域
发布于
2024年8月4日
许可协议