分享

基于SparkSQL的电影分析项目实战【含代码】

问题导读

1.本项目需求有哪些?
2.类的入口是什么?
3.各个需求是如何实现的?

代码下载:

项目介绍
数据集介绍
使用MovieLens的名称为ml-25m.zip的数据集,使用的文件时movies.csv和ratings.csv,上述文件的下载地址为:

百度网盘下载:
链接: https://pan.baidu.com/s/1gcWegpzozkkAjZE9l2xD0Q 提取码: ijqd

  • movies.csv
该文件是电影数据,对应的为维表数据,大小为2.89MB,包括6万多部电影,其数据格式为[movieId,title,genres],分别对应**[电影id,电影名称,电影所属分类]**,样例数据如下所示:逗号分隔
  1. 1,Toy Story (1995),Adventure|Animation|Children|Comedy|Fantasy
复制代码

  • ratings.csv
该文件为定影评分数据,对应为事实表数据,大小为646MB,其数据格式为:[userId,movieId,rating,timestamp],分别对应**[用户id,电影id,评分,时间戳]**,样例数据如下所示:逗号分隔
  1. 1,296,5,1147880044
复制代码

项目代码结构
aa1.png

需求分析
  • 需求1:查找电影评分个数超过5000,且平均评分较高的前十部电影名称及其对应的平均评分
  • 需求2:查找每个电影类别及其对应的平均评分
  • 需求3:查找被评分次数较多的前十部电影

代码讲解
  • DemoMainApp
该类是程序执行的入口,主要是获取数据源,转换成DataFrame,并调用封装好的业务逻辑类。
  1. object DemoMainApp {
  2.   // 文件路径
  3.   private val MOVIES_CSV_FILE_PATH = "file:///e:/movies.csv"
  4.   private val RATINGS_CSV_FILE_PATH = "file:///e:/ratings.csv"
  5.   def main(args: Array[String]): Unit = {
  6.     // 创建spark session
  7.     val spark = SparkSession
  8.       .builder
  9.       .master("local[4]")
  10.       .getOrCreate
  11.     // schema信息
  12.     val schemaLoader = new SchemaLoader
  13.     // 读取Movie数据集
  14.     val movieDF = readCsvIntoDataSet(spark, MOVIES_CSV_FILE_PATH, schemaLoader.getMovieSchema)
  15.     // 读取Rating数据集
  16.     val ratingDF = readCsvIntoDataSet(spark, RATINGS_CSV_FILE_PATH, schemaLoader.getRatingSchema)
  17.     // 需求1:查找电影评分个数超过5000,且平均评分较高的前十部电影名称及其对应的平均评分
  18.     val bestFilmsByOverallRating = new BestFilmsByOverallRating
  19.     //bestFilmsByOverallRating.run(movieDF, ratingDF, spark)
  20.     // 需求2:查找每个电影类别及其对应的平均评分
  21.     val genresByAverageRating = new GenresByAverageRating
  22.     //genresByAverageRating.run(movieDF, ratingDF, spark)
  23.     // 需求3:查找被评分次数较多的前十部电影
  24.     val mostRatedFilms = new MostRatedFilms
  25.     mostRatedFilms.run(movieDF, ratingDF, spark)
  26.     spark.close()
  27.   }
  28.   /**
  29.     * 读取数据文件,转成DataFrame
  30.     *
  31.     * @param spark
  32.     * @param path
  33.     * @param schema
  34.     * @return
  35.     */
  36.   def readCsvIntoDataSet(spark: SparkSession, path: String, schema: StructType) = {
  37.     val dataSet = spark.read
  38.       .format("csv")
  39.       .option("header", "true")
  40.       .schema(schema)
  41.       .load(path)
  42.     dataSet
  43.   }
  44. }
复制代码


  • Entry
该类为实体类,封装了数据源的样例类和结果表的样例类
  1. class Entry {
  2. }
  3. case class Movies(
  4.                    movieId: String, // 电影的id
  5.                    title: String, // 电影的标题
  6.                    genres: String // 电影类别
  7.                  )
  8. case class Ratings(
  9.                     userId: String, // 用户的id
  10.                     movieId: String, // 电影的id
  11.                     rating: String, // 用户评分
  12.                     timestamp: String // 时间戳
  13.                   )
  14. // 需求1MySQL结果表
  15. case class tenGreatestMoviesByAverageRating(
  16.                                              movieId: String, // 电影的id
  17.                                              title: String, // 电影的标题
  18.                                              avgRating: String // 电影平均评分
  19.                                            )
  20. // 需求2MySQL结果表
  21. case class topGenresByAverageRating(
  22.                                      genres: String, //电影类别
  23.                                      avgRating: String // 平均评分
  24.                                    )
  25. // 需求3MySQL结果表
  26. case class tenMostRatedFilms(
  27.                               movieId: String, // 电影的id
  28.                               title: String, // 电影的标题
  29.                               ratingCnt: String // 电影被评分的次数
  30.                             )
  31. SchemaLoader
  32. 该类封装了数据集的schema信息,主要用于读取数据源是指定schema信息
  33. class SchemaLoader {
  34.   // movies数据集schema信息
  35.   private val movieSchema = new StructType()
  36.     .add("movieId", DataTypes.StringType, false)
  37.     .add("title", DataTypes.StringType, false)
  38.     .add("genres", DataTypes.StringType, false)
  39. // ratings数据集schema信息
  40.   private val ratingSchema = new StructType()
  41.     .add("userId", DataTypes.StringType, false)
  42.     .add("movieId", DataTypes.StringType, false)
  43.     .add("rating", DataTypes.StringType, false)
  44.     .add("timestamp", DataTypes.StringType, false)
  45.   def getMovieSchema: StructType = movieSchema
  46.   def getRatingSchema: StructType = ratingSchema
  47. }
复制代码

  • JDBCUtil
该类封装了连接MySQL的逻辑,主要用于连接MySQL,在业务逻辑代码中会使用该工具类获取MySQL连接,将结果数据写入到MySQL中。
  1. object JDBCUtil {
  2.   val dataSource = new ComboPooledDataSource()
  3.   val user = "root"
  4.   val password = "123qwe"
  5.   val url = "jdbc:mysql://localhost:3306/mydb"
  6.   dataSource.setUser(user)
  7.   dataSource.setPassword(password)
  8.   dataSource.setDriverClass("com.mysql.jdbc.Driver")
  9.   dataSource.setJdbcUrl(url)
  10.   dataSource.setAutoCommitOnClose(false)
  11. // 获取连接
  12.   def getQueryRunner(): Option[QueryRunner]={
  13.     try {
  14.       Some(new QueryRunner(dataSource))
  15.     }catch {
  16.       case e:Exception =>
  17.         e.printStackTrace()
  18.         None
  19.     }
  20.   }
  21. }
复制代码

需求1实现
  • BestFilmsByOverallRating
需求1实现的业务逻辑封装。该类有一个run()方法,主要是封装计算逻辑。
  1. /**
  2.   * 需求1:查找电影评分个数超过5000,且平均评分较高的前十部电影名称及其对应的平均评分
  3.   */
  4. class BestFilmsByOverallRating extends Serializable {
  5.   def run(moviesDataset: DataFrame, ratingsDataset: DataFrame, spark: SparkSession) = {
  6.     import spark.implicits._
  7.     // 将moviesDataset注册成表
  8.     moviesDataset.createOrReplaceTempView("movies")
  9.     // 将ratingsDataset注册成表
  10.     ratingsDataset.createOrReplaceTempView("ratings")
  11.     // 查询SQL语句
  12.     val ressql1 =
  13.       """
  14.          |WITH ratings_filter_cnt AS (
  15.          |SELECT
  16.          |            movieId,
  17.          |            count( * ) AS rating_cnt,
  18.          |            avg( rating ) AS avg_rating
  19.          |FROM
  20.          |            ratings
  21.          |GROUP BY
  22.          |            movieId
  23.          |HAVING
  24.          |            count( * ) >= 5000
  25.          |),
  26.          |ratings_filter_score AS (
  27.          |SELECT
  28.          |     movieId, -- 电影id
  29.          |     avg_rating -- 电影平均评分
  30.          |FROM ratings_filter_cnt
  31.          |ORDER BY avg_rating DESC -- 平均评分降序排序
  32.          |LIMIT 10 -- 平均分较高的前十部电影
  33.          |)
  34.          |SELECT
  35.          |           m.movieId,
  36.          |           m.title,
  37.          |           r.avg_rating AS avgRating
  38.          |FROM
  39.          |          ratings_filter_score r
  40.          |JOIN movies m ON m.movieId = r.movieId
  41.       """.stripMargin
  42.     val resultDS = spark.sql(ressql1).as[tenGreatestMoviesByAverageRating]
  43.     // 打印数据
  44.     resultDS.show(10)
  45.     resultDS.printSchema()
  46.     // 写入MySQL
  47.     resultDS.foreachPartition(par => par.foreach(insert2Mysql(_)))
  48.   }
  49.   /**
  50.     * 获取连接,调用写入MySQL数据的方法
  51.     *
  52.     * @param res
  53.     */
  54.   private def insert2Mysql(res: tenGreatestMoviesByAverageRating): Unit = {
  55.     lazy val conn = JDBCUtil.getQueryRunner()
  56.     conn match {
  57.       case Some(connection) => {
  58.         upsert(res, connection)
  59.       }
  60.       case None => {
  61.         println("Mysql连接失败")
  62.         System.exit(-1)
  63.       }
  64.     }
  65.   }
  66.   /**
  67.     * 封装将结果写入MySQL的方法
  68.     * 执行写入操作
  69.     *
  70.     * @param r
  71.     * @param conn
  72.     */
  73.   private def upsert(r: tenGreatestMoviesByAverageRating, conn: QueryRunner): Unit = {
  74.     try {
  75.       val sql =
  76.         s"""
  77.            |REPLACE INTO `ten_movies_averagerating`(
  78.            |movieId,
  79.            |title,
  80.            |avgRating
  81.            |)
  82.            |VALUES
  83.            |(?,?,?)
  84.        """.stripMargin
  85.       // 执行insert操作
  86.       conn.update(
  87.         sql,
  88.         r.movieId,
  89.         r.title,
  90.         r.avgRating
  91.       )
  92.     } catch {
  93.       case e: Exception => {
  94.         e.printStackTrace()
  95.         System.exit(-1)
  96.       }
  97.     }
  98.   }
  99. }
复制代码

需求1结果
  • 结果表建表语句
  1. CREATE TABLE `ten_movies_averagerating` (
  2.   `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  3.   `movieId` int(11) NOT NULL COMMENT '电影id',
  4.   `title` varchar(100) NOT NULL COMMENT '电影名称',
  5.   `avgRating` decimal(10,2) NOT NULL COMMENT '平均评分',
  6.   `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
  7.   PRIMARY KEY (`id`),
  8.   UNIQUE KEY `movie_id_UNIQUE` (`movieId`)
  9. ) ENGINE=InnoDB  DEFAULT CHARSET=utf8;
复制代码

  • 统计结果
平均评分最高的前十部电影如下:
movieId
title
avgRating
318
Shawshank Redemption, The (1994)
4.41
858
Godfather, The (1972)
4.32
50
Usual Suspects, The (1995)
4.28
1221
Godfather: Part II, The (1974)
4.26
527
Schindler's List (1993)
4.25
2019
Seven Samurai (Shichinin no samurai) (1954)
4.25
904
Rear Window (1954)
4.24
1203
12 Angry Men (1957)
4.24
2959
Fight Club (1999)
4.23
1193
One Flew Over the Cuckoo's Nest (1975)
4.22

上述电影评分对应的电影中文名称为:
英文名称
中文名称
Shawshank Redemption, The (1994)
肖申克的救赎
Godfather, The (1972)
教父1
Usual Suspects, The (1995)
非常嫌疑犯
Godfather: Part II, The (1974)
教父2
Schindler's List (1993)
辛德勒的名单
Seven Samurai (Shichinin no samurai) (1954)
七武士
Rear Window (1954)
后窗
12 Angry Men (1957)
十二怒汉
Fight Club (1999)
搏击俱乐部
One Flew Over the Cuckoo's Nest (1975)
飞越疯人院

需求2实现
  • GenresByAverageRating
需求2实现的业务逻辑封装。该类有一个run()方法,主要是封装计算逻辑。
  1. **
  2.   * 需求2:查找每个电影类别及其对应的平均评分
  3.   */
  4. class GenresByAverageRating extends Serializable {
  5.   def run(moviesDataset: DataFrame, ratingsDataset: DataFrame, spark: SparkSession) = {
  6.     import spark.implicits._
  7.     // 将moviesDataset注册成表
  8.     moviesDataset.createOrReplaceTempView("movies")
  9.     // 将ratingsDataset注册成表
  10.     ratingsDataset.createOrReplaceTempView("ratings")
  11.     val ressql2 =
  12.       """
  13.         |WITH explode_movies AS (
  14.         |SELECT
  15.         |        movieId,
  16.         |        title,
  17.         |        category
  18.         |FROM
  19.         |        movies lateral VIEW explode ( split ( genres, "\\|" ) ) temp AS category
  20.         |)
  21.         |SELECT
  22.         |        m.category AS genres,
  23.         |        avg( r.rating ) AS avgRating
  24.         |FROM
  25.         |        explode_movies m
  26.         |        JOIN ratings r ON m.movieId = r.movieId
  27.         |GROUP BY
  28.         |        m.category
  29.         | """.stripMargin
  30.     val resultDS = spark.sql(ressql2).as[topGenresByAverageRating]
  31.     // 打印数据
  32.     resultDS.show(10)
  33.     resultDS.printSchema()
  34.     // 写入MySQL
  35.     resultDS.foreachPartition(par => par.foreach(insert2Mysql(_)))
  36.   }
  37.   /**
  38.     * 获取连接,调用写入MySQL数据的方法
  39.     *
  40.     * @param res
  41.     */
  42.   private def insert2Mysql(res: topGenresByAverageRating): Unit = {
  43.     lazy val conn = JDBCUtil.getQueryRunner()
  44.     conn match {
  45.       case Some(connection) => {
  46.         upsert(res, connection)
  47.       }
  48.       case None => {
  49.         println("Mysql连接失败")
  50.         System.exit(-1)
  51.       }
  52.     }
  53.   }
  54.   /**
  55.     * 封装将结果写入MySQL的方法
  56.     * 执行写入操作
  57.     *
  58.     * @param r
  59.     * @param conn
  60.     */
  61.   private def upsert(r: topGenresByAverageRating, conn: QueryRunner): Unit = {
  62.     try {
  63.       val sql =
  64.         s"""
  65.            |REPLACE INTO `genres_average_rating`(
  66.            |genres,
  67.            |avgRating
  68.            |)
  69.            |VALUES
  70.            |(?,?)
  71.        """.stripMargin
  72.       // 执行insert操作
  73.       conn.update(
  74.         sql,
  75.         r.genres,
  76.         r.avgRating
  77.       )
  78.     } catch {
  79.       case e: Exception => {
  80.         e.printStackTrace()
  81.         System.exit(-1)
  82.       }
  83.     }
  84.   }
  85. }
复制代码


需求2结果
  • 结果表建表语句
  1. CREATE TABLE genres_average_rating (
  2.     `id` INT ( 11 ) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  3.     `genres` VARCHAR ( 100 ) NOT NULL COMMENT '电影类别',
  4.     `avgRating` DECIMAL ( 10, 2 ) NOT NULL COMMENT '电影类别平均评分',
  5.     `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
  6. PRIMARY KEY ( `id` ),
  7. UNIQUE KEY `genres_UNIQUE` ( `genres` )
  8. ) ENGINE = INNODB DEFAULT CHARSET = utf8;
复制代码

  • 统计结果
共有20个电影分类,每个电影分类的平均评分为:
genres
avgRating
Film-Noir
3.93
War
3.79
Documentary
3.71
Crime
3.69
Drama
3.68
Mystery
3.67
Animation
3.61
IMAX
3.6
Western
3.59
Musical
3.55
Romance
3.54
Adventure
3.52
Thriller
3.52
Fantasy
3.51
Sci-Fi
3.48
Action
3.47
Children
3.43
Comedy
3.42
(no genres listed)
3.33
Horror
3.29
电影分类对应的中文名称为:
分类
中文名称
Film-Noir
黑色电影
War
战争
Documentary
纪录片
Crime
犯罪
Drama
历史剧
Mystery
推理
Animation
动画片
IMAX
巨幕电影
Western
西部电影
Musical
音乐
Romance
浪漫
Adventure
冒险
Thriller
惊悚片
Fantasy
魔幻电影
Sci-Fi
科幻
Action
动作
Children
儿童
Comedy
喜剧
(no genres listed)
未分类
Horror
恐怖

需求3实现
  • MostRatedFilms
    需求3实现的业务逻辑封装。该类有一个run()方法,主要是封装计算逻辑。
  1. /**
  2.   * 需求3:查找被评分次数较多的前十部电影.
  3.   */
  4. class MostRatedFilms extends Serializable {
  5.    def run(moviesDataset: DataFrame, ratingsDataset: DataFrame,spark: SparkSession) = {
  6.      import spark.implicits._
  7.      // 将moviesDataset注册成表
  8.      moviesDataset.createOrReplaceTempView("movies")
  9.      // 将ratingsDataset注册成表
  10.      ratingsDataset.createOrReplaceTempView("ratings")
  11. val ressql3 =
  12.   """
  13.     |WITH rating_group AS (
  14.     |    SELECT
  15.     |       movieId,
  16.     |       count( * ) AS ratingCnt
  17.     |    FROM ratings
  18.     |    GROUP BY movieId
  19.     |),
  20.     |rating_filter AS (
  21.     |    SELECT
  22.     |       movieId,
  23.     |       ratingCnt
  24.     |    FROM rating_group
  25.     |    ORDER BY ratingCnt DESC
  26.     |    LIMIT 10
  27.     |)
  28.     |SELECT
  29.     |    m.movieId,
  30.     |    m.title,
  31.     |    r.ratingCnt
  32.     |FROM
  33.     |    rating_filter r
  34.     |JOIN movies m ON r.movieId = m.movieId
  35.     |
  36.   """.stripMargin
  37.      val resultDS = spark.sql(ressql3).as[tenMostRatedFilms]
  38.      // 打印数据
  39.      resultDS.show(10)
  40.      resultDS.printSchema()
  41.      // 写入MySQL
  42.      resultDS.foreachPartition(par => par.foreach(insert2Mysql(_)))
  43.   }
  44.   /**
  45.     * 获取连接,调用写入MySQL数据的方法
  46.     *
  47.     * @param res
  48.     */
  49.   private def insert2Mysql(res: tenMostRatedFilms): Unit = {
  50.     lazy val conn = JDBCUtil.getQueryRunner()
  51.     conn match {
  52.       case Some(connection) => {
  53.         upsert(res, connection)
  54.       }
  55.       case None => {
  56.         println("Mysql连接失败")
  57.         System.exit(-1)
  58.       }
  59.     }
  60.   }
  61.   /**
  62.     * 封装将结果写入MySQL的方法
  63.     * 执行写入操作
  64.     *
  65.     * @param r
  66.     * @param conn
  67.     */
  68.   private def upsert(r: tenMostRatedFilms, conn: QueryRunner): Unit = {
  69.     try {
  70.       val sql =
  71.         s"""
  72.            |REPLACE INTO `ten_most_rated_films`(
  73.            |movieId,
  74.            |title,
  75.            |ratingCnt
  76.            |)
  77.            |VALUES
  78.            |(?,?,?)
  79.        """.stripMargin
  80.       // 执行insert操作
  81.       conn.update(
  82.         sql,
  83.         r.movieId,
  84.         r.title,
  85.         r.ratingCnt
  86.       )
  87.     } catch {
  88.       case e: Exception => {
  89.         e.printStackTrace()
  90.         System.exit(-1)
  91.       }
  92.     }
  93.   }
  94. }
复制代码

需求3结果
  • 结果表创建语句
  1. CREATE TABLE ten_most_rated_films (
  2.     `id` INT ( 11 ) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  3.     `movieId` INT ( 11 ) NOT NULL COMMENT '电影Id',
  4.     `title` varchar(100) NOT NULL COMMENT '电影名称',
  5.     `ratingCnt` INT(11) NOT NULL COMMENT '电影被评分的次数',
  6.     `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
  7. PRIMARY KEY ( `id` ),
  8. UNIQUE KEY `movie_id_UNIQUE` ( `movieId` )
  9. ) ENGINE = INNODB DEFAULT CHARSET = utf8;
复制代码

  • 统计结果
movieId
title
ratingCnt
356
Forrest Gump (1994)
81491
318
Shawshank Redemption, The (1994)
81482
296
Pulp Fiction (1994)
79672
593
Silence of the Lambs, The (1991)
74127
2571
Matrix, The (1999)
72674
260
Star Wars: Episode IV - A New Hope (1977)
68717
480
Jurassic Park (1993)
64144
527
Schindler's List (1993)
60411
110
Braveheart (1995)
59184
2959
Fight Club (1999)
58773

评分次数较多的电影对应的中文名称为:
英文名称
中文名称
Forrest Gump (1994)
阿甘正传
Shawshank Redemption, The (1994)
肖申克的救赎
Pulp Fiction (1994)
低俗小说
Silence of the Lambs, The (1991)
沉默的羔羊
Matrix, The (1999)
黑客帝国
Star Wars: Episode IV - A New Hope (1977)
星球大战
Jurassic Park (1993)
侏罗纪公园
Schindler's List (1993)
辛德勒的名单
Braveheart (1995)
勇敢的心
Fight Club (1999)
搏击俱乐部

最新经典文章,欢迎关注公众号



原文链接:https://github.com/jiamx/spark_project_practise


没找到任何评论,期待你打破沉寂

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条