基于sparksql的电影分析项目实战(代码片段)

在本篇分享中，将介绍一个完整的项目案例，该案例会真实还原企业中SparkSQL的开发流程，手把手教你构建一个基于SparkSQL的分析系统。为了讲解方便，我会对代码进行拆解，完整的代码已上传至GitHub，想看完整代码可以去clone，顺便给个**Star**。以下是全文，希望本文对你有所帮助。看完记得三连：分享、点赞、在看

https://github.com/jiamx/spark_project_practise

项目介绍

数据集介绍

使用MovieLens的名称为ml-25m.zip的数据集，使用的文件时movies.csv和ratings.csv，上述文件的下载地址为：

http://files.grouplens.org/datasets/movielens/ml-25m.zip

• movies.csv

该文件是电影数据，对应的为维表数据，大小为2.89MB，包括6万多部电影，其数据格式为[movieId,title,genres],分别对应[电影id，电影名称，电影所属分类]，样例数据如下所示：逗号分隔

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

• ratings.csv

该文件为定影评分数据，对应为事实表数据，大小为646MB，其数据格式为:[userId,movieId,rating,timestamp],分别对应[用户id，电影id，评分，时间戳]，样例数据如下所示：逗号分隔

1,296,5,1147880044

项目代码结构

需求分析

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

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

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

代码讲解

• DemoMainApp

该类是程序执行的入口，主要是获取数据源，转换成DataFrame，并调用封装好的业务逻辑类。

object DemoMainApp 
  // 文件路径
  private val MOVIES_CSV_FILE_PATH = "file:///e:/movies.csv"
  private val RATINGS_CSV_FILE_PATH = "file:///e:/ratings.csv"

  def main(args: Array[String]): Unit = 
    // 创建spark session
    val spark = SparkSession
      .builder
      .master("local[4]")
      .getOrCreate
    // schema信息
    val schemaLoader = new SchemaLoader
    // 读取Movie数据集
    val movieDF = readCsvIntoDataSet(spark, MOVIES_CSV_FILE_PATH, schemaLoader.getMovieSchema)
    // 读取Rating数据集
    val ratingDF = readCsvIntoDataSet(spark, RATINGS_CSV_FILE_PATH, schemaLoader.getRatingSchema)

    // 需求1：查找电影评分个数超过5000,且平均评分较高的前十部电影名称及其对应的平均评分
    val bestFilmsByOverallRating = new BestFilmsByOverallRating
    //bestFilmsByOverallRating.run(movieDF, ratingDF, spark)

    // 需求2：查找每个电影类别及其对应的平均评分
    val genresByAverageRating = new GenresByAverageRating
    //genresByAverageRating.run(movieDF, ratingDF, spark)

    // 需求3：查找被评分次数较多的前十部电影
    val mostRatedFilms = new MostRatedFilms
    mostRatedFilms.run(movieDF, ratingDF, spark)

    spark.close()

  
  /**
    * 读取数据文件，转成DataFrame
    *
    * @param spark
    * @param path
    * @param schema
    * @return
    */
  def readCsvIntoDataSet(spark: SparkSession, path: String, schema: StructType) = 

    val dataSet = spark.read
      .format("csv")
      .option("header", "true")
      .schema(schema)
      .load(path)
    dataSet
  

• Entry

该类为实体类，封装了数据源的样例类和结果表的样例类

class 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
                                             title: String, // 电影的标题
                                             avgRating: String // 电影平均评分
                                           )

// 需求2MySQL结果表
case class topGenresByAverageRating(
                                     genres: String, //电影类别
                                     avgRating: String // 平均评分
                                   )

// 需求3MySQL结果表
case class tenMostRatedFilms(
                              movieId: String, // 电影的id
                              title: String, // 电影的标题
                              ratingCnt: String // 电影被评分的次数
                            )

• SchemaLoader

该类封装了数据集的schema信息，主要用于读取数据源是指定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

• JDBCUtil

该类封装了连接MySQL的逻辑，主要用于连接MySQL，在业务逻辑代码中会使用该工具类获取MySQL连接，将结果数据写入到MySQL中。

object JDBCUtil 
  val dataSource = new ComboPooledDataSource()
  val user = "root"
  val password = "123qwe"
  val url = "jdbc:mysql://localhost:3306/mydb"

  dataSource.setUser(user)
  dataSource.setPassword(password)
  dataSource.setDriverClass("com.mysql.jdbc.Driver")
  dataSource.setJdbcUrl(url)
  dataSource.setAutoCommitOnClose(false)
// 获取连接
  def getQueryRunner(): Option[QueryRunner]=
    try 
      Some(new QueryRunner(dataSource))
    catch 
      case e:Exception =>
        e.printStackTrace()
        None
    
  

需求1实现

• BestFilmsByOverallRating

需求1实现的业务逻辑封装。该类有一个run()方法，主要是封装计算逻辑。

/**
  * 需求1：查找电影评分个数超过5000,且平均评分较高的前十部电影名称及其对应的平均评分
  */
class BestFilmsByOverallRating extends Serializable 

  def run(moviesDataset: DataFrame, ratingsDataset: DataFrame, spark: SparkSession) = 
    import spark.implicits._

    // 将moviesDataset注册成表
    moviesDataset.createOrReplaceTempView("movies")
    // 将ratingsDataset注册成表
    ratingsDataset.createOrReplaceTempView("ratings")

    // 查询SQL语句
    val ressql1 =
      """
         |WITH ratings_filter_cnt AS (
         |SELECT
         |     movieId,
         |     count( * ) AS rating_cnt,
         |     avg( rating ) AS avg_rating
         |FROM
         |     ratings
         |GROUP BY
         |     movieId
         |HAVING
         |     count( * ) >= 5000
         |),
         |ratings_filter_score AS (
         |SELECT
         |     movieId, -- 电影id
         |     avg_rating -- 电影平均评分
         |FROM ratings_filter_cnt
         |ORDER BY avg_rating DESC -- 平均评分降序排序
         |LIMIT 10 -- 平均分较高的前十部电影
         |)
         |SELECT
         |    m.movieId,
         |    m.title,
         |    r.avg_rating AS avgRating
         |FROM
         |   ratings_filter_score r
         |JOIN movies m ON m.movieId = r.movieId
      """.stripMargin

    val resultDS = spark.sql(ressql1).as[tenGreatestMoviesByAverageRating]
    // 打印数据
    resultDS.show(10)
    resultDS.printSchema()
    // 写入MySQL
    resultDS.foreachPartition(par => par.foreach(insert2Mysql(_)))
  

  /**
    * 获取连接，调用写入MySQL数据的方法
    *
    * @param res
    */
  private def insert2Mysql(res: tenGreatestMoviesByAverageRating): Unit = 
    lazy val conn = JDBCUtil.getQueryRunner()
    conn match 
      case Some(connection) => 
        upsert(res, connection)
      
      case None => 
        println("Mysql连接失败")
        System.exit(-1)
      
    
  

  /**
    * 封装将结果写入MySQL的方法
    * 执行写入操作
    *
    * @param r
    * @param conn
    */
  private def upsert(r: tenGreatestMoviesByAverageRating, conn: QueryRunner): Unit = 
    try 
      val sql =
        s"""
           |REPLACE INTO `ten_movies_averagerating`(
           |movieId,
           |title,
           |avgRating
           |)
           |VALUES
           |(?,?,?)
       """.stripMargin
      // 执行insert操作
      conn.update(
        sql,
        r.movieId,
        r.title,
        r.avgRating
      )
     catch 
      case e: Exception => 
        e.printStackTrace()
        System.exit(-1)
      
    
  

需求1结果

• 结果表建表语句

CREATE TABLE `ten_movies_averagerating` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `movieId` int(11) NOT NULL COMMENT '电影id',
  `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  DEFAULT CHARSET=utf8;

• 统计结果

平均评分最高的前十部电影如下：

上述电影评分对应的电影中文名称为：

需求2实现

• GenresByAverageRating

需求2实现的业务逻辑封装。该类有一个run()方法，主要是封装计算逻辑。

**
  * 需求2：查找每个电影类别及其对应的平均评分
  */
class GenresByAverageRating extends Serializable 
  def run(moviesDataset: DataFrame, ratingsDataset: DataFrame, spark: SparkSession) = 
    import spark.implicits._
    // 将moviesDataset注册成表
    moviesDataset.createOrReplaceTempView("movies")
    // 将ratingsDataset注册成表
    ratingsDataset.createOrReplaceTempView("ratings")

    val ressql2 =
      """
        |WITH explode_movies AS (
        |SELECT
        | movieId,
        | title,
        | category
        |FROM
        | movies lateral VIEW explode ( split ( genres, "\\\\|" ) ) temp AS category
        |)
        |SELECT
        | m.category AS genres,
        | avg( r.rating ) AS avgRating
        |FROM
        | explode_movies m
        | JOIN ratings r ON m.movieId = r.movieId
        |GROUP BY
        | m.category
        | """.stripMargin

    val resultDS = spark.sql(ressql2).as[topGenresByAverageRating]

    // 打印数据
    resultDS.show(10)
    resultDS.printSchema()
    // 写入MySQL
    resultDS.foreachPartition(par => par.foreach(insert2Mysql(_)))

  

  /**
    * 获取连接，调用写入MySQL数据的方法
    *
    * @param res
    */
  private def insert2Mysql(res: topGenresByAverageRating): Unit = 
    lazy val conn = JDBCUtil.getQueryRunner()
    conn match 
      case Some(connection) => 
        upsert(res, connection)
      
      case None => 
        println("Mysql连接失败")
        System.exit(-1)
      
    
  

  /**
    * 封装将结果写入MySQL的方法
    * 执行写入操作
    *
    * @param r
    * @param conn
    */
  private def upsert(r: topGenresByAverageRating, conn: QueryRunner): Unit = 
    try 
      val sql =
        s"""
           |REPLACE INTO `genres_average_rating`(
           |genres,
           |avgRating
           |)
           |VALUES
           |(?,?)
       """.stripMargin
      // 执行insert操作
      conn.update(
        sql,
        r.genres,
        r.avgRating
      )
     catch 
      case e: Exception => 
        e.printStackTrace()
        System.exit(-1)
      
    
  

需求2结果

• 结果表建表语句

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 DEFAULT CHARSET = utf8;

• 统计结果

共有20个电影分类，每个电影分类的平均评分为：

电影分类对应的中文名称为：

需求3实现

• MostRatedFilms

  需求3实现的业务逻辑封装。该类有一个run()方法，主要是封装计算逻辑。

/**
  * 需求3：查找被评分次数较多的前十部电影.
  */
class MostRatedFilms extends Serializable 
   def run(moviesDataset: DataFrame, ratingsDataset: DataFrame,spark: SparkSession) = 

     import spark.implicits._

     // 将moviesDataset注册成表
     moviesDataset.createOrReplaceTempView("movies")
     // 将ratingsDataset注册成表
     ratingsDataset.createOrReplaceTempView("ratings")

val ressql3 =
  """
    |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
    |
  """.stripMargin

     val resultDS = spark.sql(ressql3).as[tenMostRatedFilms]
     // 打印数据
     resultDS.show(10)
     resultDS.printSchema()
     // 写入MySQL
     resultDS.foreachPartition(par => par.foreach(insert2Mysql(_)))

  

  /**
    * 获取连接，调用写入MySQL数据的方法
    *
    * @param res
    */
  private def insert2Mysql(res: tenMostRatedFilms): Unit = 
    lazy val conn = JDBCUtil.getQueryRunner()
    conn match 
      case Some(connection) => 
        upsert(res, connection)
      
      case None => 
        println("Mysql连接失败")
        System.exit(-1)
      
    
  

  /**
    * 封装将结果写入MySQL的方法
    * 执行写入操作
    *
    * @param r
    * @param conn
    */
  private def upsert(r: tenMostRatedFilms, conn: QueryRunner): Unit = 
    try 
      val sql =
        s"""
           |REPLACE INTO `ten_most_rated_films`(
           |movieId,
           |title,
           |ratingCnt
           |)
           |VALUES
           |(?,?,?)
       """.stripMargin
      // 执行insert操作
      conn.update(
        sql,
        r.movieId,
        r.title,
        r.ratingCnt
      )
     catch 
      case e: Exception => 
        e.printStackTrace()
        System.exit(-1)
      
    
  

需求3结果

• 结果表创建语句

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` ),
UNIQUE KEY `movie_id_UNIQUE` ( `movieId` )
) ENGINE = INNODB DEFAULT CHARSET = utf8;

• 统计结果

评分次数较多的电影对应的中文名称为：

总结

本文主要是基于SparkSQL对MovieLens数据集进行统计分析，完整实现了三个需求，并给对每个需求都给出了详细的代码实现和结果分析。本案例还原了企业使用SparkSQL进行实现数据统计的基本流程，通过本文，或许你对SparkSQL的应用有了更加深刻的认识，希望本文对你有所帮助。