Spark中RDDDataFrame和DataSet的区别与联系

Posted 2023-02-14 YaoYong_BigData

一、RDD、DataFrame和DataSet的定义

        在开始Spark RDD与DataFrame与Dataset之间的比较之前，先让我们看一下Spark中的RDD，DataFrame和Datasets的定义：

Spark RDD：RDD代表弹性分布式数据集。它是记录的只读分区集合。 RDD是Spark的基本数据结构。它允许程序员以容错方式在大型集群上执行内存计算。

Spark Dataframe：与RDD不同，数据以列的形式组织起来，类似于关系数据库中的表。它是一个不可变的分布式数据集合。 Spark中的DataFrame允许开发人员将数据结构(类型)加到分布式数据集合上，从而实现更高级别的抽象。

Spark Dataset：Apache Spark中的Dataset是DataFrame API的扩展，它提供了类型安全(type-safe)，面向对象(object-oriented)的编程接口。 Dataset利用Catalyst optimizer可以让用户通过类似于sql的表达式对数据进行查询。

1. 细说DataFrame

        DataFrame的前身是SchemaRDD。Spark1.3更名为DataFrame。不继承RDD，自己实现了RDD的大部分功能。
与RDD类似，DataFrame也是一个分布式数据集：
1）DataFrame可以看做分布式 Row 对象的集合，提供了由列组成的详细模式信息，使其可以得到优化。DataFrame 不仅有比RDD更多的算子，还可以进行执行计划的优化。
2）DataFrame更像传统数据库的二维表格，除了数据以外，还记录数据的结构信息，即schema。
3）DataFrame也支持嵌套数据类型（struct、array和map）。
4）DataFrame API提供的是一套高层的关系操作，比函数式的RDD API要更加友好，门槛更低。
5）Dataframe的劣势在于在编译期缺少类型安全检查，导致运行时出错。

2. 细说DataSet

1）DataSet是在Spark1.6中添加的新的接口。
2）与RDD相比，保存了更多的描述信息，概念上等同于关系型数据库中的二维表。
3）与DataFrame相比，保存了类型信息，是强类型的，提供了编译时类型检查。
4）调用Dataset的方法先会生成逻辑计划，然后Spark的优化器进行优化，最终生成物理计划，然后提交到集群中运行。
5）DataSet包含了DataFrame的功能，在Spark2.0中两者得到了统一：DataFrame表示为DataSet[Row]，即DataSet的子集。

3. 结构图解：

 1）RDD[Person]：
以Person 为类型参数，但不了解其内部结构。
2）DataFrame：
提供了详细的结构信息schema 列的名称和类型。这样看起来就像一张表了。
3）DataSet：
不光有schema 信息，还有类型信息。

4. 数据图解：

假设RDD中的两行数据长这样：RDD[Person]：

 那么DataFrame中的数据长这样：

DataFrame = RDD[Row] + Schema；DataFrame 的前身是 SchemaRDD。

 那么Dataset中的数据长这样：Dataset[Person] = DataFrame + 泛型

 或者长这样（每行数据是个Object）：Dataset[Row]，即DataFrame = DataSet[Row]

        DataSet包含了DataFrame的功能，Spark2.0中两者统一，DataFrame表示为DataSet[Row]，即DataSet的子集。

5. 补充说明:Row & Schema

        Row是一个泛化的无类型 JVM object，Row 对象表示的是一个行，Row 的操作类似于 Scala 中的 Map 数据类型。

// 一个对象就是一个对象
val p = People(name = "zhangsan", age = 10)

// 同样一个对象, 还可以通过一个 Row 对象来表示
val row = Row("zhangsan", 10)

// 获取 Row 中的内容
println(row.get(1))
println(row(1))

// 获取时可以指定类型
println(row.getAs[Int](1))

// 同时 Row 也是一个样例类, 可以进行 match
row match 
  case Row(name, age) => println(name, age)

什么是schema？
        DataFrame中提供了详细的数据结构信息，从而使得SparkSQL可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么，DataFrame中的数据结构信息，即为schema。

二、三者的共性

1. RDD、DataFrame、DataSet全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利;

2. 三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算;

3. 三者有许多共同的函数，如filter，排序等;

4. 在对DataFrame和Dataset进行操作许多操作都需要这个包:import spark.implicits._（在创建好SparkSession对象后尽量直接导入）；

5. 三者都会根据 Spark 的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出；

6. 三者都有partition的概念；

7. DataFrame和Dataset均可使用模式匹配获取各个字段的值和类型。

DataFrame:

testDF.map
      case Row(col1:String,col2:Int)=>
        println(col1);println(col2)
        col1
      case _=>
        ""
    

Dataset：

case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型
    testDS.map
      case Coltest(col1:String,col2:Int)=>
        println(col1);println(col2)
        col1
      case _=>
        ""
    

三、RDD、DataFrame和DataSet的联系

1. RDD

优点:

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

缺点:

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

2. 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也不是面向对象风格的。

        DataFrame也可以叫Dataset[Row],每一行的类型是Row，不解析，每一行究竟有哪些字段，各个字段又是什么类型都无从得知，只能用上面提到的getAS方法或者共性中的第七条提到的模式匹配拿出特定字段。

优点:

        DataFrame 内部有明确 Scheme 结构，即列名、列字段类型都是已知的，这带来的好处是可以减少数据读取以及更好地优化执行计划，从而保证查询效率。

缺点：

(1)Dataframe的劣势在于在编译期缺少类型安全检查，导致运行时出错。

(2)DataFrame虽然是结构化的，但是其所含的值并没有对应一个class，所以spark就定义了一个class名为Row，作为DataFrame的数据的数据结构。所以DataFrame等价于Dataset[Row]。但是Row又没有定义field，具体包含哪些字段，没法直接取出来，所以只能通过Row的各种方法比如getAs[Int](xxx)来获取属性xxx的内容。而Dataset每一行是什么类型是不一定的，在自定义了case class之后可以很自由的获得每一行的信息。所以DataFrame在获取内部数据的时候，方法数据的属性没有Dataset方便。

3. DataSet

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

        当序列化数据时, Encoder产生字节码与off-heap进行交互, 能够达到按需访问数据的效果, 而不用反序列化整个对象。

四、DataFrame和DataSet的区别

第一点: DataFrame 表达的含义是一个支持函数式操作的 表, 而 Dataset 表达是是一个类似 RDD 的东西, Dataset 可以处理任何对象。

第二点: DataFrame 中所存放的是 Row 对象, 而 Dataset 中可以存放任何类型的对象。

val spark: SparkSession = new sql.SparkSession.Builder()
  .appName("hello")
  .master("local[6]")
  .getOrCreate()

import spark.implicits._

val df: DataFrame = Seq(People("zhangsan", 15), People("lisi", 15)).toDF()       

val ds: Dataset[People] = Seq(People("zhangsan", 15), People("lisi", 15)).toDS() 

DataFrame 就是 Dataset[Row]
Dataset 的范型可以是任意类型
​第三点: DataFrame 的操作方式和 Dataset 是一样的, 但是对于强类型操作而言, 它们处理的类型不同。

DataFrame 在进行强类型操作时候, 例如 map 算子, 其所处理的数据类型永远是 Row：

df.map( (row: Row) => Row(row.get(0), row.getAs[Int](1) * 10) )(RowEncoder.apply(df.schema)).show()

但是对于 Dataset 来讲, 其中是什么类型, 它就处理什么类型：

ds.map( (item: People) => People(item.name, item.age * 10) ).show()

第四点: DataFrame 只能做到运行时类型检查, Dataset 能做到编译和运行时都有类型检查。

1. DataFrame 中存放的数据以 Row 表示, 一个 Row 代表一行数据, 这和关系型数据库类似；

2. DataFrame 在进行 map 等操作的时候, DataFrame 不能直接使用 Person 这样的 Scala 对象, 所以无法做到编译时检查；

3. Dataset 表示的具体的某一类对象, 例如 Person, 所以再进行 map 等操作的时候, 传入的是具体的某个 Scala 对象, 如果调用错了方法, 编译时就会被检查出来。

val ds: Dataset[People] = Seq(People("zhangsan", 15), People("lisi", 15)).toDS()
//这行代码明显报错, 无法通过编译
ds.map(person => person.hello) 

五、​​​​​​​三者的互相转化