面试官嫌我Sql写的太low？要求我重写还加了三个需求？——二战Spark电影评分数据分析

CREATE DATABASE db_movies;
USE db_movies;
CREATE TABLE `ten_movies_avgrating` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `movieId` int(11) NOT NULL COMMENT '电影id',
  `ratingNum` int(11) NOT NULL COMMENT '评分个数',
  `title` varchar(100) NOT NULL COMMENT '电影名称',
  `avgRating` decimal(10,2) NOT NULL COMMENT '平均评分',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `movie_id_UNIQUE` (`movieId`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
CREATE TABLE `genres_average_rating` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `genres` varchar(100) NOT NULL COMMENT '电影类别',
  `avgRating` decimal(10,2) NOT NULL COMMENT '电影类别平均评分',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `genres_UNIQUE` (`genres`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
CREATE TABLE `ten_most_rated_films` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `movieId` int(11) NOT NULL COMMENT '电影Id',
  `title` varchar(100) NOT NULL COMMENT '电影名称',
  `ratingCnt` int(11) NOT NULL COMMENT '电影被评分的次数',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

package cn.movies.Packet
/**
  * @author ChinaManor
  *         #Description Entry
  *         #Date: 6/6/2021 17:23
  */
object Entry {
  case class Movies(
                     movieId: String, // 电影的id
                     title: String, // 电影的标题
                     genres: String // 电影类别
                   )
  case class Ratings(
                      userId: String, // 用户的id
                      movieId: String, // 电影的id
                      rating: String, // 用户评分
                      timestamp: String // 时间戳
                    )
  // 需求1MySQL结果表
  case class tenGreatestMoviesByAverageRating(
                                               movieId: String, // 电影的id
                                               ratingNum:String,
                                               title: String, // 电影的标题
                                               avgRating: String // 电影平均评分
                                             )
  // 需求2MySQL结果表
  case class topGenresByAverageRating(
                                       genres: String, //电影类别
                                       avgRating: String // 平均评分
                                     )
  // 需求3MySQL结果表
  case class tenMostRatedFilms(
                                movieId: String, // 电影的id
                                title: String, // 电影的标题
                                ratingCnt: String // 电影被评分的次数
                              )
}

package cn.movies.Packet
import org.apache.spark.sql.types.{DataTypes, StructType}
/**
  * @author ChinaManor
  *         #Description Schema
  *         #Date: 6/6/2021 17:34
  */
object Schema {
  class SchemaLoader {
    // movies数据集schema信息
    private val movieSchema = new StructType()
      .add("movieId", DataTypes.StringType, false)
      .add("title", DataTypes.StringType, false)
      .add("genres", DataTypes.StringType, false)
    // ratings数据集schema信息
    private val ratingSchema = new StructType()
      .add("userId", DataTypes.StringType, false)
      .add("movieId", DataTypes.StringType, false)
      .add("rating", DataTypes.StringType, false)
      .add("timestamp", DataTypes.StringType, false)
    def getMovieSchema: StructType = movieSchema
    def getRatingSchema: StructType = ratingSchema
  }
}

// 创建spark session
    val spark = SparkSession
      .builder
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[4]")
      .getOrCreate

val schemaLoader = new SchemaLoader

// 文件路径
  private val MOVIES_CSV_FILE_PATH = "D:\\Users\\Administrator\\Desktop\\exam0601\\datas\\movies.csv"
  private val RATINGS_CSV_FILE_PATH = "D:\\Users\\Administrator\\Desktop\\exam0601\\datas\\ratings.csv"
  /**
    * 读取数据文件，转成DataFrame
    *
    * @param spark
    * @param path
    * @param schema
    * @return
    */
  def readCsvIntoDataSet(spark: SparkSession, path: String, schema: StructType) = {
    val dataDF: DataFrame = spark.read
      .format("csv")
      .option("header", "true")
      .schema(schema)
      .load(path)
    dataDF
  }

WITH ratings_filter_cnt AS (
SELECT
     movieId,
     count( * ) AS rating_cnt,
     Round(avg( rating ),2) AS avg_rating
FROM
     ratings
GROUP BY
     movieId
HAVING
     count( * ) >= 50
),
ratings_filter_score AS (
SELECT
     movieId, -- 电影id
     rating_cnt, -- 个数
     avg_rating -- 电影平均评分
FROM ratings_filter_cnt
ORDER BY avg_rating DESC -- 平均评分降序排序
LIMIT 10 -- 平均分较高的前十部电影
)
SELECT
    m.movieId,
    r.rating_cnt AS ratingNum,
    m.title,
    r.avg_rating AS avgRating
FROM
   ratings_filter_score r
JOIN movies m ON m.movieId = r.movieId ORDER BY r.avg_rating DESC

def saveToMysql(reportDF: DataFrame) = {
    // TODO: 使用SparkSQL提供内置Jdbc数据源保存数据
    reportDF
      .coalesce(1)
      .write
      // 追加模式，将数据追加到MySQL表中，再次运行，主键存在，报错异常
      .mode(SaveMode.Append)
      // 覆盖模式，无需测试，直接将以前数据全部删除，再次重新重建表，肯定不行
      //.mode(SaveMode.Overwrite)
      .format("jdbc")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("url", "jdbc:mysql://192.168.88.100:3306/db_movies?serverTimezone=UTC&characterEncoding=utf8&useUnicode=true")
      .option("user", "root")
      .option("password", "123456")
      .option("dbtable", "db_movies.ten_most_rated_films")
      .save()
  }

// 需求2：查找每个电影类别及其对应的平均评分
WITH explode_movies AS (
SELECT
 movieId,
 title,
 category
FROM
 movies lateral VIEW explode ( split ( genres, "\\|" ) ) temp AS category  //爆炸函数拆一下｜ 
)
SELECT
 m.category AS genres,
 Round(avg( r.rating ),2) AS avgRating
FROM
 explode_movies m
 JOIN ratings r ON m.movieId = r.movieId
GROUP BY
 m.category
ORDER BY avgRating DESC
   // 需求3：查找被评分次数较多的前十部电影
WITH rating_group AS (
    SELECT
       movieId,
       count( * ) AS ratingCnt
    FROM ratings
    GROUP BY movieId
),
rating_filter AS (
    SELECT
       movieId,
       ratingCnt
    FROM rating_group
    ORDER BY ratingCnt DESC
    LIMIT 10
)
SELECT
    m.movieId,
    m.title,
    r.ratingCnt
FROM
    rating_filter r
JOIN movies m ON r.movieId = m.movieId ORDER BY r.ratingCnt DESC