import java.sql.{Connection, DriverManager, PreparedStatement} import org.apache.spark.sql.{DataFrame, Row, SparkSession} import org.apache.spark.sql.functions._ import org.apache.spark.storage.StorageLevel /** * 电影评分数据分析,需求如下: * 需求1:查找电影评分个数超过50,且平均评分较高的前十部电影名称及其对应的平均评分 * 电影ID 评分个数 电影名称 平均评分 更新时间 * movie_id、rating_num、title、rating_avg、update_time * 需求2:查找每个电影类别及其对应的平均评分 * 电影类别 电影类别平均评分 更新时间 * genre、 rating_avg 、update_time * 需求3:查找被评分次数较多的前十部电影 * 电影ID 电影名称 电影被评分的次数 更新时间 * movie_id、title、rating_num、 update_time */ object MetricsAppMain { // 文件路径 private val RATINGS_CSV_FILE_PATH = "J:\\t4\\FlinkCommodityRecommendationSystem-main\\FlinkCommodityRecommendationSystem-main\\recommendation\\src\\main\\resources\\ratings.csv" // private val MOVIES_CSV_FILE_PATH = "D:\\Users\\Administrator\\Desktop\\exam0601\\datas\\movies.csv" def main(args: Array[String]): Unit = { // step1、创建SparkSession实例对象 val spark: SparkSession = createSparkSession(this.getClass) import spark.implicits._ /* 分析需求可知,三个需求最终结果,需要使用事实表数据和维度表数据关联,所以先数据拉宽,再指标计算 TODO: 按照数据仓库分层理论管理数据和开发指标 - 第一层(最底层):ODS层 直接加CSV文件数据为DataFrame - 第二层(中间层):DW层 将加载业务数据(电影评分数据)和维度数据(电影基本信息数据)进行Join关联,拉宽操作 - 第三层(最上层):DA层/APP层 依据需求开发程序,计算指标,进行存储到MySQL表 */ // step2、【ODS层】:加载数据,CSV格式数据,文件首行为列名称 val ratingDF: DataFrame = readCsvFile(spark, RATINGS_CSV_FILE_PATH, verbose = false) // val movieDF: DataFrame = readCsvFile(spark, MOVIES_CSV_FILE_PATH, verbose = false) // step3、【DW层】:将电影评分数据与电影信息数据进行关联,数据拉宽操作 // val detailDF: DataFrame = joinDetail(ratingDF, movieDF) printConsole(ratingDF) // step4、【DA层】:按照业务需求,进行指标统计分析 computeMetric(ratingDF) Thread.sleep(1000000) // 应用结束,关闭资源 spark.stop() } /** * 构建SparkSession实例对象,默认情况下本地模式运行 */ def createSparkSession(clazz: Class[_], master: String = "local[2]"): SparkSession = { SparkSession.builder() .appName(clazz.getSimpleName.stripSuffix("$")) .master(master) .config("spark.sql.shuffle.partitions", "2") .getOrCreate() } /** * 读取CSV格式文本文件数据,封装到DataFrame数据集 */ def readCsvFile(spark: SparkSession, path: String, verbose: Boolean = true): DataFrame = { val dataframe: DataFrame = spark.read // 设置分隔符为逗号 .option("sep", ",") // 文件首行为列名称 .option("header", "true") // 依据数值自动推断数据类型 .option("inferSchema", "true") .csv(path) if(verbose){ printConsole(dataframe) } // 返回数据集 dataframe } /** * 按照业务需求,进行指标统计,默认情况下,结果数据打印控制台 */ def computeMetric(dataframe: DataFrame): Unit = { // TODO: 缓存数据 dataframe.persist(StorageLevel.MEMORY_AND_DISK) // 需求1:查找电影评分个数超过50,且平均评分较高的前十部电影名称及其对应的平均评分 val top10FilesDF: DataFrame = top10Films(dataframe) //printConsole(top10FilesDF) upsertToMySQL( top10FilesDF, // "replace into test.rating (id, userId, productId, score, timestamp) values (null, ?, ?, ?, ?)", // (pstmt: PreparedStatement, row: Row) => { pstmt.setInt(1, row.getAs[Int]("userId")) pstmt.setInt(2, row.getAs[Int]("productId")) pstmt.setDouble(3, row.getAs[Double]("score")) pstmt.setInt(4, row.getAs[Int]("timestamp")) } ) // 释放资源 dataframe.unpersist() } /** * 需求:查找电影评分个数超过50,且平均评分较高的前十部电影名称及其对应的平均评分 * 电影ID 评分个数 电影名称 平均评分 更新时间 * movie_id、rating_num、title、rating_avg、update_time */ def top10Films(dataframe: DataFrame): DataFrame = { import dataframe.sparkSession.implicits._ dataframe // 添加日期字段 // .withColumn("update_time", current_timestamp()) } /** * 将DataFrame数据集打印控制台,显示Schema信息和前10条数据 */ def printConsole(dataframe: DataFrame): Unit = { // 显示Schema信息 dataframe.printSchema() // 显示前10条数据 dataframe.show(10, truncate = false) } /** * 将数据保存至MySQL表中,采用replace方式,当主键存在时,更新数据;不存在时,插入数据 * @param dataframe 数据集 * @param sql 插入数据SQL语句 * @param accept 函数,如何设置Row中每列数据到SQL语句中占位符值 */ def upsertToMySQL(dataframe: DataFrame, sql: String, accept: (PreparedStatement, Row) => Unit): Unit = { // 降低分区数目,对每个分区进行操作 dataframe.coalesce(1).foreachPartition{iter => // step1. 加载驱动类 Class.forName("com.mysql.cj.jdbc.Driver") // 声明变量 var conn: Connection = null var pstmt: PreparedStatement = null try{ // step2. 创建连接 conn = DriverManager.getConnection( "jdbc:mysql://120.26.162.238:33306/?serverTimezone=UTC&characterEncoding=utf8&useUnicode=true", "root", "123456" ) pstmt = conn.prepareStatement(sql) // step3. 插入数据 iter.foreach{row => // 设置SQL语句中占位符的值 accept(pstmt, row) // 加入批次中 pstmt.addBatch() } // 批量执行批次 pstmt.executeBatch() }catch { case e: Exception => e.printStackTrace() }finally { // step4. 关闭连接 if(null != pstmt) pstmt.close() if(null != conn) conn.close() } } } }