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

本文涉及的产品
云数据库 RDS MySQL Serverless,0.5-2RCU 50GB
简介: 面试官嫌我Sql写的太low?要求我重写还加了三个需求?——二战Spark电影评分数据分析

引言

大家好,我是ChinaManor,直译过来就是中国码农的意思,俺希望自己能成为国家复兴道路的铺路人,大数据领域的耕耘者,平凡但不甘于平庸的人。


Spark综合练习——电影评分数据分析

这是我的上篇博文,当时仅是做了一个实现案例(demo级别 ),没想到居然让我押中了题,还让我稳稳的及格了(这次测试试卷难度极大,考60分都能在班上排进前10

不过我在复盘的时候,发现自己的致命弱点:写sql的能力太菜了。。

于是我重做了一遍,并满足了导师提的3个需求:

需求1: 查找电影评分个数超过50,且平均评分较高的前十部电影名称及其对应的平均评分

需求2: 查找每个电影类别及其对应的平均评分

需求3: 查找被评分次数较多的前十部电影

数据介绍:使用的文件movies.csv和ratings.csv

movies.csv该文件是电影数据,对应的为维表数据,其数据格式为

movieId title genres

电影id 电影名称 电影所属分类

样例数据如下所示:逗号分隔

1,Toy Story (1995),Adventure|Animation|Children|Comedy|Fantasy

ratings.csv该文件为定影评分数据,其数据格式为

userId movieId rating timestamp

电影id 电影名称 电影所属分类 时间戳

建表语句

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;

项目结构一览图

由题意可知

先创建实体类,字段是从建表语句中得来的。


Entry.scala

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 // 电影被评分的次数
                              )
}

再创建个表结构~~

Schema.scala

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
  }
}

然后开始写Main方法,其实只有区区八十行代码。。。

spark总要有实例对象吧。

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

然后 new个schema信息

val schemaLoader = new SchemaLoader

然后尝试读取csv文件,

// 读取Movie数据集

val movieDF: DataFrame = readCsvIntoDataSet(spark, MOVIES_CSV_FILE_PATH, schemaLoader.getMovieSchema)

// 读取Rating数据集

val ratingDF: DataFrame = readCsvIntoDataSet(spark, RATINGS_CSV_FILE_PATH, schemaLoader.getRatingSchema)

发现读取方法和路径都没有,于是补救一下

// 文件路径
  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
  }

紧接着重头戏来了。。

写sql语句,在大数据行业懂得写sql就等于会了80%


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

关键点在于

WITH XXX AS SELECT


最后保存写入mysql表中

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()
  }

另外两个需求的SQL:

// 需求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

总结

以上便是spark电影评分数据分析二次改写,比之前一篇sql更复杂,需求更多,

希望今晚的考试顺利通关@~@

如果需要完整版的代码可以私信我获取

愿你读过之后有自己的收获,如果有收获不妨一键三连一下~


相关实践学习
基于CentOS快速搭建LAMP环境
本教程介绍如何搭建LAMP环境,其中LAMP分别代表Linux、Apache、MySQL和PHP。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
相关文章
|
10天前
|
SQL 自然语言处理 数据挖掘
大模型与数据分析:探索Text-to-SQL(上)
大模型与数据分析:探索Text-to-SQL(上)
33 0
|
10天前
|
SQL 自然语言处理 数据挖掘
大模型与数据分析:探索Text-to-SQL(中)
大模型与数据分析:探索Text-to-SQL(中)
35 0
|
10天前
|
SQL 存储 数据挖掘
大模型与数据分析:探索Text-to-SQL(下)
大模型与数据分析:探索Text-to-SQL(下)
35 3
|
1月前
|
SQL 数据库 C#
C# .NET面试系列十一:数据库SQL查询(附建表语句)
#### 第1题 用一条 SQL 语句 查询出每门课都大于80 分的学生姓名 建表语句: ```sql create table tableA ( name varchar(10), kecheng varchar(10), fenshu int(11) ) DEFAULT CHARSET = 'utf8'; ``` 插入数据 ```sql insert into tableA values ('张三', '语文', 81); insert into tableA values ('张三', '数学', 75); insert into tableA values ('李四',
65 2
C# .NET面试系列十一:数据库SQL查询(附建表语句)
|
2月前
|
SQL 存储 数据库
面试题19: 如何优化SQL查询?
面试题19: 如何优化SQL查询?
面试题19: 如何优化SQL查询?
|
1月前
|
存储 关系型数据库 MySQL
最全MySQL面试60题(含答案):存储引擎+数据库锁+索引+SQL优化等
最全MySQL面试60题(含答案):存储引擎+数据库锁+索引+SQL优化等
169 0
|
26天前
|
SQL 关系型数据库 MySQL
MySQL SQL语句面试准备
MySQL SQL语句面试准备
12 0
|
2月前
|
SQL 关系型数据库 MySQL
|
3月前
|
SQL 数据挖掘 数据库
SQL数据分析实战:从导入到高级查询的完整指南
SQL数据分析实战:从导入到高级查询的完整指南
59 0
|
3月前
|
SQL
面试必备杀技:SQL查询专项训练(二)
面试必备杀技:SQL查询专项训练

热门文章

最新文章