你好,游客 登录
背景:
阅读新闻

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

[日期:2018-04-26] 来源:csdn  作者: a11a22334455 [字体: ]

RDD

优点:

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

缺点:

  1. 序列化和反序列化的性能开销 
    无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化.
  2. 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()




  }
}

收藏 推荐 打印 | 阅读: