零、本节学习目标

利用RDD计算总分与平均分
利用RDD统计每日新增用户
利用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.day07

import 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)
    // 读取成绩文件，生成RDD
    val 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))
    })
    // 基于二元组成绩列表创建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")
    // 关闭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.day07

import 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)
    // 读取文件，得到RDD
    val 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()
    // 取分组后的日期集合最小值，计数为1
    val rdd4 = rdd3.map(line => (line._2.min, 1))
    // 按键计数，得到每日新增用户数
    val result = rdd4.countByKey()
    // 让统计结果按日期升序
    val keys = result.keys.toList.sorted
    keys.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._1
          val top3 = item._2.toList.sortWith(_ > _).take(3)
          (name, top3)
        })
top3.collect.foreach(println)

执行上述代码

（5）按指定格式输出结果

top3.collect.foreach(line => {
      val name = line._1
      var 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.day07

import 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._1
        val 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）运行程序，查看结果