RDD

优点:

编译时类型安全
编译时就能检查出类型错误
面向对象的编程风格
直接通过类名点的方式来操作数据

缺点:

序列化和反序列化的性能开销
无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化.
GC的性能开销
频繁的创建和销毁对象, 势必会增加GC

DataFrame

DataFrame引入了schema和off-heap

schema : RDD每一行的数据, 结构都是一样的. 这个结构就存储在schema中. Spark通过schame就能够读懂数据, 因此在通信和IO时就只需要序列化和反序列化数据, 而结构的部分就可以省略了.
off-heap : 意味着JVM堆以外的内存, 这些内存直接受操作系统管理（而不是JVM）。Spark能够以二进制的形式序列化数据(不包括结构)到off-heap中, 当要操作数据时, 就直接操作off-heap内存. 由于Spark理解schema, 所以知道该如何操作.

off-heap就像地盘, schema就像地图, Spark有地图又有自己地盘了, 就可以自己说了算了, 不再受JVM的限制, 也就不再收GC的困扰了.

通过schema和off-heap, DataFrame解决了RDD的缺点, 但是却丢了RDD的优点. DataFrame不是类型安全的, API也不是面向对象风格的.

DataSet

DataSet结合了RDD和DataFrame的优点, 并带来的一个新的概念Encoder

当序列化数据时, Encoder产生字节码与off-heap进行交互, 能够达到按需访问数据的效果, 而不用反序列化整个对象. Spark还没有提供自定义Encoder的API, 但是未来会加入.

package com.dt.spark.sparksql

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}

/**
* 电影点评系统用户行为分析：用户观看电影和点评电影的所有行为数据的采集、过滤、处理和展示：
* 数据采集：企业中一般越来越多的喜欢直接把Server中的数据发送给Kafka，因为更加具备实时性；
* 数据过滤：趋势是直接在Server端进行数据过滤和格式化，当然采用Spark SQL进行数据的过滤也是一种主要形式；
* 数据处理：
* 1，一个基本的技巧是，先使用传统的SQL去实现一个下数据处理的业务逻辑（自己可以手动模拟一些数据）；
* 2，再一次推荐使用DataSet去实现业务功能尤其是统计分析功能；
* 3，如果你想成为专家级别的顶级Spark人才，请使用RDD实现业务功能，为什么？运行的时候是基于RDD的！
*
* 数据：强烈建议大家使用Parquet
* 1，"ratings.dat"：UserID::MovieID::Rating::Timestamp
* 2，"users.dat"：UserID::Gender::Age::OccupationID::Zip-code
* 3，"movies.dat"：MovieID::Title::Genres
* 4, "occupations.dat"：OccupationID::OccupationName 一般情况下都会以程序中数据结构Haskset的方式存在，是为了做mapjoin
*/
object Movie_Users_Analyzer_DateSet {

case class User(UserID:String, Gender:String, Age:String, OccupationID:String, Zip_Code:String)
case class Rating(UserID:String, MovieID:String, Rating:Double, Timestamp:String)
case class Movie(MovieID:String, Title:String, Genres:String)

def main(args: Array[String]){

Logger.getLogger("org").setLevel(Level.ERROR)

var masterUrl = "local[8]" //默认程序运行在本地Local模式中，主要学习和测试；
var dataPath = "moviedata/medium/" //数据存放的目录；

/**
* 当我们把程序打包运行在集群上的时候一般都会传入集群的URL信息，在这里我们假设如果传入
* 参数的话，第一个参数只传入Spark集群的URL第二个参数传入的是数据的地址信息；
*/
if(args.length > 0) {
masterUrl = args(0)
} else if (args.length > 1) {
dataPath = args(1)
}

/**
* 创建Spark会话上下文SparkSession和集群上下文SparkContext，在SparkConf中可以进行各种依赖和参数的设置等，
* 大家可以通过SparkSubmit脚本的help去看设置信息，其中SparkSession统一了Spark SQL运行的不同环境。
*/
val sparkConf = new SparkConf().setMaster(masterUrl).setAppName("Movie_Users_Analyzer_DataSet")

/**
* SparkSession统一了Spark SQL执行时候的不同的上下文环境，也就是说Spark SQL无论运行在那种环境下我们都可以只使用
* SparkSession这样一个统一的编程入口来处理DataFrame和DataSet编程，不需要关注底层是否有Hive等。
*/
val spark = SparkSession
.builder()
.config(sparkConf)
.getOrCreate()

val sc = spark.sparkContext //从SparkSession获得的上下文，这是因为我们读原生文件的时候或者实现一些Spark SQL目前还不支持的功能的时候需要使用SparkContext

import spark.implicits._
/**
* 读取数据，用什么方式读取数据呢？在这里是使用RDD!
*/
val usersRDD = sc.textFile(dataPath + "users.dat")
val moviesRDD = sc.textFile(dataPath + "movies.dat")
val occupationsRDD = sc.textFile(dataPath + "occupations.dat")
val ratingsRDD = sc.textFile(dataPath + "ratings.dat")

/**
* 功能一：通过DataFrame实现某特定电影观看者中男性和女性不同年龄分别有多少人？
* 1，从点评数据中获得观看者的信息ID；
* 2，把ratings和users表进行join操作获得用户的性别信息；
* 3，使用内置函数（内部包含超过200个内置函数）进行信息统计和分析
* 在这里我们通过DataFrame来实现：首先通过DataFrame的方式来表现ratings和users的数据，然后进行join和统计操作
*/

println("功能一：通过DataFrame实现某特定电影观看者中男性和女性不同年龄分别有多少人？")
val schemaforusers = StructType("UserID::Gender::Age::OccupationID::Zip_Code".split("::").
map(column => StructField(column, StringType, true))) //使用Struct方式把Users的数据格式化,即在RDD的基础上增加数据的元数据信息
val usersRDDRows = usersRDD.map(_.split("::")).map(line => Row(line(0).trim,line(1).
trim,line(2).trim,line(3).trim,line(4).trim)) //把我们的每一条数据变成以Row为单位的数据
val usersDataFrame = spark.createDataFrame(usersRDDRows, schemaforusers) //结合Row和StructType的元数据信息基于RDD创建DataFrame，这个时候RDD就有了元数据信息的描述
val usersDataSet = usersDataFrame.as[User]

val schemaforratings = StructType("UserID::MovieID".split("::").
map(column => StructField(column, StringType, true))).
add("Rating", DoubleType, true).
add("Timestamp",StringType, true)

val ratingsRDDRows = ratingsRDD.map(_.split("::")).map(line => Row(line(0).trim,line(1).
trim,line(2).trim.toDouble,line(3).trim))
val ratingsDataFrame = spark.createDataFrame(ratingsRDDRows, schemaforratings)
val ratingsDataSet = ratingsDataFrame.as[Rating]

val schemaformovies = StructType("MovieID::Title::Genres".split("::").
map(column => StructField(column, StringType, true))) //使用Struct方式把Users的数据格式化,即在RDD的基础上增加数据的元数据信息
val moviesRDDRows = moviesRDD.map(_.split("::")).map(line => Row(line(0).trim,line(1).
trim,line(2).trim)) //把我们的每一条数据变成以Row为单位的数据
val moviesDataFrame = spark.createDataFrame(moviesRDDRows, schemaformovies) //结合Row和StructType的元数据信息基于RDD创建DataFrame，这个时候RDD就有了元数据信息的描述
val moviesDataSet = moviesDataFrame.as[Movie]

println
ratingsDataFrame.filter(s" MovieID = 1193") //这里能够直接指定MovieID的原因是DataFrame中有该元数据信息！
.join(usersDataFrame, "UserID") //Join的时候直接指定基于UserID进行Join，这相对于原生的RDD操作而言更加方便快捷
.select("Gender", "Age") //直接通过元数据信息中的Gender和Age进行数据的筛选
.groupBy("Gender", "Age") //直接通过元数据信息中的Gender和Age进行数据的groupBy操作
.count() //基于groupBy分组信息进行count统计操作
.show(10) //显示出分组统计后的前10条信息
println("功能一：通过DataSet实现某特定电影观看者中男性和女性不同年龄分别有多少人？")
ratingsDataSet.filter(s" MovieID = 1193") //这里能够直接指定MovieID的原因是DataFrame中有该元数据信息！
.join(usersDataFrame, "UserID") //Join的时候直接指定基于UserID进行Join，这相对于原生的RDD操作而言更加方便快捷
.select("Gender", "Age") //直接通过元数据信息中的Gender和Age进行数据的筛选
.groupBy("Gender", "Age") //直接通过元数据信息中的Gender和Age进行数据的groupBy操作
.count() //基于groupBy分组信息进行count统计操作
.show(10) //显示出分组统计后的前10条信息
/**
* 功能二：用SQL语句实现某特定电影观看者中男性和女性不同年龄分别有多少人？
* 1，注册临时表，写SQL语句需要Table；
* 2，基于上述注册的零时表写SQL语句；
*/
println("功能二：用GlobalTempView的SQL语句实现某特定电影观看者中男性和女性不同年龄分别有多少人？")
ratingsDataFrame.createGlobalTempView("ratings")
usersDataFrame.createGlobalTempView("users")

spark.sql("SELECT Gender, Age, count(*) from global_temp.users u join global_temp.ratings as r on u.UserID = r.UserID where MovieID = 1193" +
" group by Gender, Age").show(10)

println("功能二：用LocalTempView的SQL语句实现某特定电影观看者中男性和女性不同年龄分别有多少人？")
ratingsDataFrame.createTempView("ratings")
usersDataFrame.createTempView("users")

spark.sql("SELECT Gender, Age, count(*) from users u join ratings as r on u.UserID = r.UserID where MovieID = 1193" +
" group by Gender, Age").show(10)

/**
* 功能三：使用DataFrame进行电影流行度分析：所有电影中平均得分最高（口碑最好）的电影及观看人数最高的电影（流行度最高）
* "ratings.dat"：UserID::MovieID::Rating::Timestamp
* 得分最高的Top10电影实现思路：如果想算总的评分的话一般肯定需要reduceByKey操作或者aggregateByKey操作
* 第一步：把数据变成Key-Value，大家想一下在这里什么是Key，什么是Value。把MovieID设置成为Key，把Rating设置为Value；
* 第二步：通过reduceByKey操作或者aggregateByKey实现聚合，然后呢？
* 第三步：排序，如何做？进行Key和Value的交换
*/

println("通过纯粹使用DataFrame方式计算所有电影中平均得分最高（口碑最好）的电影TopN:")
ratingsDataFrame.select("MovieID", "Rating").groupBy("MovieID").
avg("Rating").orderBy($"avg(Rating)".desc).show(10)
println("通过纯粹使用DataSet方式计算所有电影中平均得分最高（口碑最好）的电影TopN:")
ratingsDataSet.select("MovieID", "Rating").groupBy("MovieID").
avg("Rating").orderBy($"avg(Rating)".desc).show(10)
/**
* 上面的功能计算的是口碑最好的电影，接下来我们分析粉丝或者观看人数最多的电影
*/

println("纯粹通过DataFrame的方式计算最流行电影即所有电影中粉丝或者观看人数最多(最流行电影)的电影TopN:")
// ratingsDataFrame.select("MovieID","Timestamp").
// ratingsDataFrame.select("MovieID").
ratingsDataFrame.groupBy("MovieID").count().
orderBy($"count".desc).show(10)

println("纯粹通过DataSet的方式计算最流行电影即所有电影中粉丝或者观看人数最多(最流行电影)的电影TopN:")
ratingsDataSet.groupBy("MovieID").count().
orderBy($"count".desc).show(10)

/**
* 功能四：分析最受男性喜爱的电影Top10和最受女性喜爱的电影Top10
* 1，"users.dat"：UserID::Gender::Age::OccupationID::Zip-code
* 2，"ratings.dat"：UserID::MovieID::Rating::Timestamp
* 分析：单单从ratings中无法计算出最受男性或者女性喜爱的电影Top10,因为该RDD中没有Gender信息，如果我们需要使用
* Gender信息来进行Gender的分类，此时一定需要聚合，当然我们力求聚合的使用是mapjoin（分布式计算的Killer
* 是数据倾斜，map端的join是一定不会数据倾斜），在这里可否使用mapjoin呢？不可以，因为用户的数据非常多！
* 所以在这里要使用正常的Join，此处的场景不会数据倾斜，因为用户一般都很均匀的分布（但是系统信息搜集端要注意黑客攻击）
*
* Tips：
* 1，因为要再次使用电影数据的RDD，所以复用了前面Cache的ratings数据
* 2, 在根据性别过滤出数据后关于TopN部分的代码直接复用前面的代码就行了。
* 3, 要进行join的话需要key-value；
* 4, 在进行join的时候时刻通过take等方法注意join后的数据格式 (3319,((3319,50,4.5),F))
* 5, 使用数据冗余来实现代码复用或者更高效的运行，这是企业级项目的一个非常重要的技巧！
*/

val genderRatingsDataFrame = ratingsDataFrame.join(usersDataFrame, "UserID").cache()
val genderRatingsDataSet = ratingsDataSet.join(usersDataSet, "UserID").cache()
val maleFilteredRatingsDataFrame = genderRatingsDataFrame.filter("Gender= 'M'").select("MovieID", "Rating")
val maleFilteredRatingsDataSet = genderRatingsDataSet.filter("Gender= 'M'").select("MovieID", "Rating")
val femaleFilteredRatingsDataFrame = genderRatingsDataFrame.filter("Gender= 'F'").select("MovieID", "Rating")
val femaleFilteredRatingsDataSet = genderRatingsDataSet.filter("Gender= 'F'").select("MovieID", "Rating")

/**
* (855,5.0)
(6075,5.0)
(1166,5.0)
(3641,5.0)
(1045,5.0)
(4136,5.0)
(2538,5.0)
(7227,5.0)
(8484,5.0)
(5599,5.0)
*/

println("纯粹使用DataFrame实现所有电影中最受男性喜爱的电影Top10:")
maleFilteredRatingsDataFrame.groupBy("MovieID").avg("Rating").orderBy($"avg(Rating)".desc).show(10)
println("纯粹使用DataSet实现所有电影中最受男性喜爱的电影Top10:")
maleFilteredRatingsDataSet.groupBy("MovieID").avg("Rating").orderBy($"avg(Rating)".desc).show(10)
/**
* (789,5.0)
(855,5.0)
(32153,5.0)
(4763,5.0)
(26246,5.0)
(2332,5.0)
(503,5.0)
(4925,5.0)
(8767,5.0)
(44657,5.0)
*/

println("纯粹使用DataFrame实现所有电影中最受女性喜爱的电影Top10:")
femaleFilteredRatingsDataFrame.groupBy("MovieID").avg("Rating").orderBy($"avg(Rating)".desc, $"MovieID".desc).show(10)

println("纯粹使用DataSet实现所有电影中最受女性喜爱的电影Top10:")
femaleFilteredRatingsDataSet.groupBy("MovieID").avg("Rating").orderBy($"avg(Rating)".desc, $"MovieID".desc).show(10)

/**
* 思考题：如果想让RDD和DataFrame计算的TopN的每次结果都一样，该如何保证？现在的情况是例如计算Top10，而其同样评分的不止10个，所以每次都会
* 从中取出10个，这就导致的大家的结果不一致，这个时候，我们可以使用一个新的列参与排序：
* 如果是RDD的话，该怎么做呢？这个时候就要进行二次排序，按照我们前面和大家讲解的二次排序的视频内容即可。
* 如果是DataFrame的话，该如何做呢？此时就非常简单，我们只需要再orderBy函数中增加一个排序维度的字段即可，简单的不可思议！
*/

/**
* 功能五：最受不同年龄段人员欢迎的电影TopN
* "users.dat"：UserID::Gender::Age::OccupationID::Zip-code
* 思路：首先还是计算TopN，但是这里的关注点有两个：
* 1，不同年龄阶段如何界定，关于这个问题其实是业务的问题，当然，你实际在实现的时候可以使用RDD的filter中的例如
* 13 < age <18,这样做会导致运行时候大量的计算，因为要进行扫描，所以会非常耗性能。所以，一般情况下，我们都是
* 在原始数据中直接对要进行分组的年龄段提前进行好ETL, 例如进行ETL后产生以下的数据：
* - Gender is denoted by a "M" for male and "F" for female
* - Age is chosen from the following ranges:
* 1: "Under 18"
* 18: "18-24"
* 25: "25-34"
* 35: "35-44"
* 45: "45-49"
* 50: "50-55"
* 56: "56+"
* 2，性能问题：
* 第一点：你实际在实现的时候可以使用RDD的filter中的例如13 < age <18,这样做会导致运行时候大量的计算，因为要进行
* 扫描，所以会非常耗性能，我们通过提前的ETL把计算发生在Spark业务逻辑运行以前，用空间换时间，当然这些实现也可以
* 使用Hive，因为Hive语法支持非常强悍且内置了最多的函数；
* 第二点：在这里要使用mapjoin，原因是targetUsers数据只有UserID，数据量一般不会太多
*/

println("纯粹通过DataFrame的方式实现所有电影中QQ或者微信核心目标用户最喜爱电影TopN分析:")

ratingsDataFrame.join(usersDataFrame, "UserID").filter("Age = '18'").groupBy("MovieID").
count().orderBy($"count".desc).printSchema()

ratingsDataSet.join(usersDataSet, "UserID").filter("Age = '18'").groupBy("MovieID").
count().orderBy($"count".desc).printSchema()

/**
* Tips:
* 1,orderBy操作需要在join之后进行
*/
println("纯粹通过DataFrame的方式实现所有电影中QQ或者微信核心目标用户最喜爱电影TopN分析:")
ratingsDataFrame.join(usersDataFrame, "UserID").filter("Age = '18'").groupBy("MovieID").
count().join(moviesDataFrame, "MovieID").select("Title", "count").orderBy($"count".desc).show(10)
println("纯粹通过DataSet的方式实现所有电影中QQ或者微信核心目标用户最喜爱电影TopN分析:")
ratingsDataSet.join(usersDataSet, "UserID").filter("Age = '18'").groupBy("MovieID").
count().join(moviesDataSet, "MovieID").select("Title", "count").sort($"count".desc).show(10)

/**
* 淘宝核心目标用户最喜爱电影TopN分析
* (Pulp Fiction (1994),959)
(Silence of the Lambs, The (1991),949)
(Forrest Gump (1994),935)
(Jurassic Park (1993),894)
(Shawshank Redemption, The (1994),859)
*/

println("纯粹通过DataFrame的方式实现所有电影中淘宝核心目标用户最喜爱电影TopN分析:")
ratingsDataFrame.join(usersDataFrame, "UserID").filter("Age = '25'").groupBy("MovieID").
count().join(moviesDataFrame, "MovieID").select("Title", "count").orderBy($"count".desc).show(10)

println("纯粹通过DataSet的方式实现所有电影中淘宝核心目标用户最喜爱电影TopN分析:")
ratingsDataSet.join(usersDataSet, "UserID").filter("Age = '25'").groupBy("MovieID").
count().join(moviesDataSet, "MovieID").select("Title", "count").sort($"count".desc).limit(10).show()

// while(true){} //和通过Spark shell运行代码可以一直看到Web终端的原理是一样的，因为Spark Shell内部有一个LOOP循环

sc.stop()

}
}

spark dataframe和dataSet用电影点评数据实战相关推荐

《Python Spark 2.0 Hadoop机器学习与大数据实战_林大贵(著)》pdf
<Python+Spark 2.0+Hadoop机器学习与大数据实战> 五星好评+强烈推荐的一本书,虽然内容可能没有很深入,但作者非常用心的把每一步操作详细的列出来并给出说明,让我们跟着做 ...
spark python教程_Python Spark 2.0 Hadoop机器学习与大数据实战完整pdf_IT教程网
资源名称:Python Spark 2.0 Hadoop机器学习与大数据实战完整pdf 第1章 Python Spark机器学习与Hadoop大数据 1 第2章 VirtualBox虚拟机软件的安装 ...
Spark性能优化 -- Spark SQL、DataFrame、Dataset
本文将详细分析和总结Spark SQL及其DataFrame.Dataset的相关原理和优化过程. Spark SQL简介 Spark SQL是Spark中具有大规模关系查询的结构化数据处理模块 ...
Spark中RDD、DataFrame和DataSet的区别与联系
一.RDD.DataFrame和DataSet的定义在开始Spark RDD与DataFrame与Dataset之间的比较之前,先让我们看一下Spark中的RDD,DataFrame和Dataset ...
Spark商业案例与性能调优实战100课》第11课：商业案例之通过纯粹通过DataFrame分析大数据电影点评系仿QQ和微信、淘宝等用户群分析与实战
Spark商业案例与性能调优实战100课>第11课:商业案例之通过纯粹通过DataFrame分析大数据电影点评系仿QQ和微信.淘宝等用户群分析与实战
基于Spark实现电影点评系统用户行为分析—DataFrame篇（二）
文章目录 1.介绍 2.业务统计 3.代码实现 1.介绍 Spark SQL有三种不同实现方式:(1)使用DataFrame与RDD结合的方式.(2)纯粹使用DataFrame的方式.(3)使用Dat ...
【大数据开发】SparkSQL——RDD、DataFrame、DataSet相互转换、DSL常用方法、SQL风格语法、Spark读写操作、获取Column对象的方式
take,takeAsList是Action操作 limit⽅法获取指定DataFrame的前n⾏记录,得到⼀个新的DataFrame对象.和take与head不同的是,limit⽅法不是Action ...
大数据入门：Spark RDD、DataFrame、DataSet
在Spark的学习当中,RDD.DataFrame.DataSet可以说都是需要着重理解的专业名词概念.尤其是在涉及到数据结构的部分,理解清楚这三者的共性与区别,非常有必要.今天的大数据入门分享,我们 ...
Spark商业案例与性能调优实战100课》第2课：商业案例之通过RDD实现分析大数据电影点评系统中电影流行度分析
Spark商业案例与性能调优实战100课>第2课:商业案例之通过RDD实现分析大数据电影点评系统中电影流行度分析 package com.dt.spark.coresimport org.apa ...

spark dataframe和dataSet用电影点评数据实战

RDD

DataFrame

DataSet

spark dataframe和dataSet用电影点评数据实战相关推荐

最新文章

热门文章