前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >2021年大数据Spark(二十八):SparkSQL案例三电影评分数据分析

2021年大数据Spark(二十八):SparkSQL案例三电影评分数据分析

作者头像
Lansonli
发布2021-10-09 16:56:13
1.3K0
发布2021-10-09 16:56:13
举报
文章被收录于专栏:Lansonli技术博客Lansonli技术博客

案例三:电影评分数据分析

     使用电影评分数据进行数据分析,分别使用DSL编程和SQL编程,熟悉数据处理函数及SQL使用,业务需求说明:

对电影评分数据进行统计分析,获取Top10电影(电影评分平均值最高,并且每个电影被评分的次数大于200)。

数据格式如下,每行数据各个字段之间使用双冒号分开:

数据处理分析步骤如下:

  1. 第一步、读取电影评分数据,从本地文件系统读取
  2.  第二步、转换数据,指定Schema信息,封装到DataFrame
  3.  第三步、基于SQL方式分析
  4.  第四步、基于DSL方式分析

代码实现

     电影评分数据分析,经过数据ETL、数据分析(SQL分析和DSL分析)及最终保存结果,整套数据处理分析流程,其中涉及到很多数据细节,完整代码如下:

代码语言:javascript
复制
package cn.itcast.sql

import java.util.Properties

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.storage.StorageLevel

/**
 * 需求:对电影评分数据进行统计分析,获取Top10电影(电影评分平均值最高,并且每个电影被评分的次数大于2000)
 */
object SparkTop10Movie {
    def main(args: Array[String]): Unit = {
        val spark = SparkSession.builder()
          .appName(this.getClass.getSimpleName.stripSuffix("$"))
          .master("local[*]")
          // TODO: 设置shuffle时分区数目
          .config("spark.sql.shuffle.partitions", "4")
          .getOrCreate()
        val sc: SparkContext = spark.sparkContext
        sc.setLogLevel("WARN")
        import spark.implicits._
        
        // 1. 读取电影评分数据,从本地文件系统读取
        val rawRatingsDS: Dataset[String] = spark.read.textFile("data/input/rating_100k.data")
        
        // 2. 转换数据
        val ratingsDF: DataFrame = rawRatingsDS
            // 过滤数据
            .filter(line => null != line && line.trim.split("\t").length == 4)
            // 提取转换数据
            .mapPartitions{iter =>
                iter.map{line =>
                    // 按照分割符分割,拆箱到变量中
                    val Array(userId, movieId, rating, timestamp) = line.trim.split("\t")
                    // 返回四元组
                    (userId, movieId, rating.toDouble, timestamp.toLong)
                }
            }
            // 指定列名添加Schema
            .toDF("userId", "movieId", "rating", "timestamp")
        /*
            root
             |-- userId: string (nullable = true)
             |-- movieId: string (nullable = true)
             |-- rating: double (nullable = false)
             |-- timestamp: long (nullable = false)
        */
        ratingsDF.printSchema()
        /*
            +------+-------+------+---------+
            |userId|movieId|rating|timestamp|
            +------+-------+------+---------+
            |     1|   1193|   5.0|978300760|
            |     1|    661|   3.0|978302109|
            |     1|    594|   4.0|978302268|
            |     1|    919|   4.0|978301368|
            +------+-------+------+---------+
         */
        ratingsDF.show(4)
        
        // TODO: 基于SQL方式分析
        // 第一步、注册DataFrame为临时视图
        ratingsDF.createOrReplaceTempView("view_temp_ratings")
        
        // 第二步、编写SQL
        val top10MovieDF: DataFrame = spark.sql(
            """
              |SELECT
              |  movieId, ROUND(AVG(rating), 2) AS avg_rating, COUNT(movieId) AS cnt_rating
              |FROM
              |  view_temp_ratings
              |GROUP BY
              |  movieId
              |HAVING
              |  cnt_rating > 200
              |ORDER BY
              |  avg_rating DESC, cnt_rating DESC
              |LIMIT
              |  10
            """.stripMargin)
        //top10MovieDF.printSchema()
        top10MovieDF.show(10, truncate = false)
        
        println("===============================================================")
        
        // TODO: 基于DSL=Domain Special Language(特定领域语言) 分析
        import org.apache.spark.sql.functions._
        val resultDF: DataFrame = ratingsDF
            // 选取字段
            .select($"movieId", $"rating")
            // 分组:按照电影ID,获取平均评分和评分次数
            .groupBy($"movieId")
            .agg(
                round(avg($"rating"), 2).as("avg_rating"),
                count($"movieId").as("cnt_rating")
            )
            // 过滤:评分次数大于200
            .filter($"cnt_rating" > 200)
            // 排序:先按照评分降序,再按照次数降序
            .orderBy($"avg_rating".desc, $"cnt_rating".desc)
            // 获取前10
            .limit(10)
        //resultDF.printSchema()
        resultDF.show(10)

        /*// TODO: 将分析的结果数据保存MySQL数据库和CSV文件
        // 结果DataFrame被使用多次,缓存
        resultDF.persist(StorageLevel.MEMORY_AND_DISK)
        
        // 1. 保存MySQL数据库表汇总
        resultDF
            .coalesce(1)
            .write
            .mode("overwrite")
            .option("driver", "com.mysql.jdbc.Driver")
            .option("user", "root")
            .option("password", "root")
            .jdbc(
                "jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8",
                "top10_movies",
                new Properties()
            )
        
        // 2. 保存CSV文件:每行数据中个字段之间使用逗号隔开
        resultDF
            .coalesce(1)
            .write.mode("overwrite")
            .csv("data/output/top10-movies")
        
        // 释放缓存数据
        resultDF.unpersist()*/

        spark.stop()
    }
}

​​​​​​​Shuffle分区数

运行上述程序时,查看WEB UI监控页面发现,某个Stage中有200个Task任务,也就是说RDD有200分区Partition。

原因:在SparkSQL中当Job中产生Shuffle时,默认的分区数(spark.sql.shuffle.partitions )为200,在实际项目中要合理的设置。可以在构建SparkSession实例对象时进行设置

代码语言:javascript
复制
val spark = SparkSession.builder()
  .appName(this.getClass.getSimpleName.stripSuffix("$"))
  .master("local[*]")
  // TODO: 设置shuffle时分区数目
  .config("spark.sql.shuffle.partitions", "4")
  .getOrCreate()
val sc: SparkContext = spark.sparkContext
sc.setLogLevel("WARN")
import spark.implicits._
本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2021-04-17 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 案例三:电影评分数据分析
  • 代码实现
  • ​​​​​​​Shuffle分区数
相关产品与服务
云数据库 MySQL
腾讯云数据库 MySQL(TencentDB for MySQL)为用户提供安全可靠,性能卓越、易于维护的企业级云数据库服务。其具备6大企业级特性,包括企业级定制内核、企业级高可用、企业级高可靠、企业级安全、企业级扩展以及企业级智能运维。通过使用腾讯云数据库 MySQL,可实现分钟级别的数据库部署、弹性扩展以及全自动化的运维管理,不仅经济实惠,而且稳定可靠,易于运维。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档