From 94ccc5c841f1f7b69ae9e3ff6eb6fd502a58a385 Mon Sep 17 00:00:00 2001 From: KennethCheng Date: Mon, 21 Sep 2026 17:00:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=E5=AE=8C=E6=88=903=E4=B8=AA=E9=9C=80?= =?UTF-8?q?=E6=B1=82=20=E9=9C=80=E6=B1=821=EF=BC=9A=E6=9F=A5=E6=89=BE?= =?UTF-8?q?=E7=94=B5=E5=BD=B1=E8=AF=84=E5=88=86=E4=B8=AA=E6=95=B0=E8=B6=85?= =?UTF-8?q?=E8=BF=875000=EF=BC=8C=E5=B9=B6=E4=B8=94=E5=B9=B3=E5=9D=87?= =?UTF-8?q?=E5=88=86=E8=BE=83=E9=AB=98=E7=9A=84=E5=89=8D=E5=8D=81=E9=83=A8?= =?UTF-8?q?=E7=94=B5=E5=BD=B1=E5=90=8D=E7=A7=B0=E5=8F=8A=E5=85=B6=E5=AF=B9?= =?UTF-8?q?=E5=BA=94=E7=9A=84=E5=B9=B3=E5=9D=87=E8=AF=84=E5=88=86=20?= =?UTF-8?q?=E9=9C=80=E6=B1=822=EF=BC=9A=E6=9F=A5=E6=89=BE=E6=AF=8F?= =?UTF-8?q?=E4=B8=AA=E7=94=B5=E5=BD=B1=E7=B1=BB=E5=88=AB=E5=8F=8A=E5=85=B6?= =?UTF-8?q?=E5=AF=B9=E5=BA=94=E7=9A=84=E5=B9=B3=E5=9D=87=E5=88=86=20?= =?UTF-8?q?=E9=9C=80=E6=B1=823=EF=BC=9A=E6=9F=A5=E6=89=BE=E8=A2=AB?= =?UTF-8?q?=E8=AF=84=E5=88=86=E6=AC=A1=E6=95=B0=E6=9C=80=E5=A4=9A=E7=9A=84?= =?UTF-8?q?=E5=89=8D=E5=8D=81=E9=83=A8=E7=94=B5=E5=BD=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/main/resources/create_table.sql | 13 ++ src/main/scala/com/example/spark/App.scala | 73 ++++++++++-- .../scala/com/example/spark/Entitys.scala | 15 ++- .../scala/com/example/utils/JDBCUtil.scala | 112 ++++++++++++++++++ 4 files changed, 200 insertions(+), 13 deletions(-) create mode 100644 src/main/scala/com/example/utils/JDBCUtil.scala diff --git a/src/main/resources/create_table.sql b/src/main/resources/create_table.sql index 90f9e79..0c15db4 100644 --- a/src/main/resources/create_table.sql +++ b/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; \ No newline at end of file diff --git a/src/main/scala/com/example/spark/App.scala b/src/main/scala/com/example/spark/App.scala index 74c06d4..3388c5f 100644 --- a/src/main/scala/com/example/spark/App.scala +++ b/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.show() +// movieData.printSchema() +// 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.show() +// ratingData.printSchema() +// 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() } } diff --git a/src/main/scala/com/example/spark/Entitys.scala b/src/main/scala/com/example/spark/Entitys.scala index ee591e8..23c6ba6 100644 --- a/src/main/scala/com/example/spark/Entitys.scala +++ b/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) \ No newline at end of file +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 // 评分次数 + ) \ No newline at end of file diff --git a/src/main/scala/com/example/utils/JDBCUtil.scala b/src/main/scala/com/example/utils/JDBCUtil.scala new file mode 100644 index 0000000..c5a48b2 --- /dev/null +++ b/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}") + } + } +} \ No newline at end of file