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