feat:完成3个需求

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

src/main/resources/create_table.sql

@@ -21,5 +21,18 @@ CREATE TABLE genres_average_rating
     `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
     PRIMARY KEY (`id`),
     UNIQUE KEY `genres_UNIQUE` (`genres`)
+) ENGINE = INNODB
+  DEFAULT CHARSET = utf8;
+
+
+CREATE TABLE movie_rating_count
+(
+    `id`          INT(11)        NOT NULL AUTO_INCREMENT COMMENT '自增id',
+    `movie_id`    INT(11)        not null comment '电影id',
+    `title`       varchar(200)   not null comment '电影名称',
+    `ratingCount`   INT(11) NOT NULL COMMENT '评分次数',
+    `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新时间',
+    PRIMARY KEY (`id`),
+    UNIQUE KEY `movie_rating_count_UNIQUE` (`movie_id`)
 ) ENGINE = INNODB
   DEFAULT CHARSET = utf8;

src/main/scala/com/example/spark/App.scala

@@ -1,5 +1,6 @@
 package com.example.spark
 
+import com.example.utils.JDBCUtil.insert2Mysql
 import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
 
 object App {
@@ -16,23 +17,23 @@ object App {
     import spark.implicits._
 
     // 1. 读取数据源
-    val movieData: Dataset[Movie] = spark.read
+    val movieData = spark.read
       .option("header","true")
       .option("inferSchema","true")
       .option("delimiter",",")
       .csv(MOVIE_DATA_PATH)
       .as[Movie]
-    movieData.printSchema()
+//    movieData.printSchema()
-    movieData.show()
+//    movieData.show()
 
-    val ratingData: Dataset[Rating] = spark.read
+    val ratingData = spark.read
       .option("header","true")
       .option("inferSchema","true")
       .option("delimiter",",")
       .csv(RATINGS_DATA_PATH)
       .as[Rating]
-    ratingData.printSchema()
+//    ratingData.printSchema()
-    ratingData.show()
+//    ratingData.show()
 
 
     // 实现3个需求
@@ -40,16 +41,64 @@ object App {
     ratingData.createOrReplaceTempView("ratings")
 
 
-
-
 //    - 需求1：查找电影评分个数超过5000，并且平均分较高的前十部电影名称及其对应的平均评分
+    val sql1 =
+      """
+        |with ratings_filter_cnt as(
+        |			select movieId,avg(rating) as avgRating
+        |				from ratings
+        |				group by movieId
+        |       having count(*) > 5000
+        |				order by avgRating desc
+        |				limit 10
+        |				)
+        |      select m.movieId,m.title,r.avgRating
+        |          from ratings_filter_cnt r
+        |          join movies m
+        |          on m.movieId = r.movieId""".stripMargin
 
-
+//    spark.sql(sql1).show()
+//        val result1 = spark.sql(sql1).as[TopTenAvgRatings]
+//        result1.foreach(insert2Mysql(_))
 //    - 需求2：查找每个电影类别及其对应的平均分
-
+    val sql2 =
+      """
+        |with explode_movies as (
+        |	select movieId,title,category
+        |		from movies lateral view explode (split(genres,"\\|")) tmp 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 result2 = spark.sql(sql2).as[TopGenresByAverageRating]
+//    result2.foreach(insert2Mysql(_))
 //    - 需求3：查找被评分次数最多的前十部电影
-
+    val sql3 =
-
+      """
+        |with movie_rating_count as (
+        |	select
+        |  max(m.movieId) movieId
+        | ,max(m.title) title
+        | ,count(1) cn
+        | from movies m
+        |  left join
+        |  ratings r
+        |  on m.movieId = r.movieId
+        |  group by m.movieId
+        |  order by  count(1) desc
+        |  limit 10
+        |)
+        |select
+        | movieId
+        |,cn ratingCount
+        |,title
+        |from
+        |movie_rating_count""".stripMargin
+      val result3 = spark.sql(sql3).as[MovieRatingCount]
+      result3.foreach(insert2Mysql(_))
+      spark.sql(sql3).show()
 
   }
 }

src/main/scala/com/example/spark/Entitys.scala

@@ -6,4 +6,17 @@ class Entitys {
 
 case class Movie(movieId: Int, title: String, genres: String)
 
-case class Rating(userId: Int, movieId: Int, rating: Double, timestamp: Long)
+case class Rating(userId: Int, movieId: Int, rating: Double, timestamp: Long)
+
+case class TopTenAvgRatings(movieId: String, title: String, avgRating: String)
+
+case class TopGenresByAverageRating(
+                                     genres: String, //电影类别
+                                     avgRating: String // 平均评分
+                                   )
+
+case class MovieRatingCount(
+                             movieId: Int, // 电影id
+                             title: String, // 电影名称
+                             ratingCount:Long // 评分次数
+                            )

src/main/scala/com/example/utils/JDBCUtil.scala

@@ -0,0 +1,112 @@
+package com.example.utils
+
+import com.example.spark.{TopGenresByAverageRating, TopTenAvgRatings, MovieRatingCount}
+import com.zaxxer.hikari.{HikariConfig, HikariDataSource}
+import org.apache.commons.dbutils.QueryRunner
+
+object JDBCUtil {
+
+  // ==================== HikariCP 配置 ====================
+
+  private val config = new HikariConfig()
+
+  private val user = "root"
+  private val password = "root123456"
+  private val url = "jdbc:mysql://192.168.48.10:3306/test"
+
+  config.setJdbcUrl(url)
+  config.setUsername(user)
+  config.setPassword(password)
+  config.setDriverClassName("com.mysql.cj.jdbc.Driver")
+
+  config.setMaximumPoolSize(10)
+  config.setMinimumIdle(5)
+  config.setIdleTimeout(30000)
+  config.setConnectionTimeout(30000)
+  config.setAutoCommit(true)
+
+  // 数据源
+  private val dataSource = new HikariDataSource(config)
+
+  // ==================== 获取 QueryRunner ====================
+
+  def getQueryRunner: Option[QueryRunner] =
+    try {
+      Some(new QueryRunner(dataSource))
+    } catch {
+      case e: Exception =>
+        e.printStackTrace()
+        None
+    }
+
+  // ==================== 写入 MySQL ====================
+
+  def insert2Mysql(res: Any): Unit =
+    getQueryRunner match {
+      case Some(queryRunner) =>
+        upsert(res, queryRunner)
+
+      case None =>
+        println("Mysql 连接失败")
+        System.exit(-1)
+    }
+
+  // ==================== 根据类型执行不同 SQL ====================
+
+  private def upsert(res: Any, queryRunner: QueryRunner): Unit = {
+    res match {
+
+      case data: TopTenAvgRatings =>
+        val sql =
+          """
+            |REPLACE INTO top_ten_avg_ratings (
+            |  movie_id,
+            |  title,
+            |  avg_rating
+            |) VALUES (?, ?, ?)
+          """.stripMargin
+
+        queryRunner.update(
+          sql,
+          data.movieId,
+          data.title,
+          data.avgRating
+        )
+
+      case data: TopGenresByAverageRating =>
+        val sql =
+          """
+            |REPLACE INTO genres_average_rating (
+            |  genres,
+            |  avg_rating
+            |) VALUES (?, ?)
+          """.stripMargin
+
+        queryRunner.update(
+          sql,
+          data.genres,
+          data.avgRating
+        )
+
+      case data: MovieRatingCount =>
+        val sql =
+          """
+            |REPLACE INTO movie_rating_count (
+            |movie_id,
+            |title,
+            |ratingCount
+            |) VALUES (?, ?, ?)
+          """.stripMargin
+
+        queryRunner.update(
+          sql,
+          Int.box(data.movieId),
+          data.title,
+          Long.box(data.ratingCount)
+        )
+
+      case _ =>
+        println(s"不支持的数据类型：${res.getClass.getName}")
+    }
+  }
+}