spark系列16：DataSet和DataFrame使用详解

妖狐艹你老母 2022-11-10 14:20 245阅读 0赞

# 1.Dataset 的特点 #

目标

1.  理解 `Dataset` 是什么
2.  理解 `Dataset` 的特性

`Dataset` 是什么?

val spark: SparkSession = new sql.SparkSession.Builder()
      .appName("hello")
      .master("local[6]")
      .getOrCreate()
    
    import spark.implicits._
    
    val dataset: Dataset[People] = spark.createDataset(Seq(People("zhangsan", 9), People("lisi", 15)))
    // 方式1: 通过对象来处理
    dataset.filter(item => item.age > 10).show()
    // 方式2: 通过字段来处理
    dataset.filter('age > 10).show()
    // 方式3: 通过类似SQL的表达式来处理
    dataset.filter("age > 10").show()

问题1: `People` 是什么?

`People` 是一个强类型的类

问题2: 这个 `Dataset` 中是结构化的数据吗?

非常明显是的, 因为 `People` 对象中有结构信息, 例如字段名和字段类型

问题3: 这个 `Dataset` 能够使用类似 `SQL` 这样声明式结构化查询语句的形式来查询吗?

当然可以, 已经演示过了

问题4: `Dataset` 是什么?

`Dataset` 是一个强类型, 并且类型安全的数据容器, 并且提供了结构化查询 `API` 和类似 `RDD` 一样的命令式 `API`

即使使用 `Dataset` 的命令式 `API`, 执行计划也依然会被优化

`Dataset` 具有 `RDD` 的方便, 同时也具有 `DataFrame` 的性能优势, 并且 `Dataset` 还是强类型的, 能做到类型安全.

scala> spark.range(1).filter('id === 0).explain(true)
    
    == Parsed Logical Plan ==
    'Filter ('id = 0)
    +- Range (0, 1, splits=8)
    
    == Analyzed Logical Plan ==
    id: bigint
    Filter (id#51L = cast(0 as bigint))
    +- Range (0, 1, splits=8)
    
    == Optimized Logical Plan ==
    Filter (id#51L = 0)
    +- Range (0, 1, splits=8)
    
    == Physical Plan ==
    *Filter (id#51L = 0)
    +- *Range (0, 1, splits=8)

`Dataset` 的底层是什么?

`Dataset` 最底层处理的是对象的序列化形式, 通过查看 `Dataset` 生成的物理执行计划, 也就是最终所处理的 `RDD`, 就可以判定 `Dataset` 底层处理的是什么形式的数据

val dataset: Dataset[People] = spark.createDataset(Seq(People("zhangsan", 9), People("lisi", 15)))
    val internalRDD: RDD[InternalRow] = dataset.queryExecution.toRdd

`dataset.queryExecution.toRdd` 这个 `API` 可以看到 `Dataset` 底层执行的 `RDD`, 这个 `RDD` 中的范型是 `InternalRow`, `InternalRow` 又称之为 `Catalyst Row`, 是 `Dataset` 底层的数据结构, 也就是说, 无论 `Dataset` 的范型是什么, 无论是 `Dataset[Person]` 还是其它的, 其最底层进行处理的数据结构都是 `InternalRow`

所以, `Dataset` 的范型对象在执行之前, 需要通过 `Encoder` 转换为 `InternalRow`, 在输入之前, 需要把 `InternalRow` 通过 `Decoder` 转换为范型对象

![cc610157b92466cac52248a8bf72b76e][]

可以获取 `Dataset` 对应的 `RDD` 表示

在 `Dataset` 中, 可以使用一个属性 `rdd` 来得到它的 `RDD` 表示, 例如 `Dataset[T] → RDD[T]`

val dataset: Dataset[People] = spark.createDataset(Seq(People("zhangsan", 9), People("lisi", 15)))
    
    /*
    (2) MapPartitionsRDD[3] at rdd at Testing.scala:159 []
     |  MapPartitionsRDD[2] at rdd at Testing.scala:159 []
     |  MapPartitionsRDD[1] at rdd at Testing.scala:159 []
     |  ParallelCollectionRDD[0] at rdd at Testing.scala:159 []
     */
    
    println(dataset.rdd.toDebugString) // 这段代码的执行计划为什么多了两个步骤?
    
    /*
    (2) MapPartitionsRDD[5] at toRdd at Testing.scala:160 []
     |  ParallelCollectionRDD[4] at toRdd at Testing.scala:160 []
     */
    
    println(dataset.queryExecution.toRdd.toDebugString)

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>使用&nbsp;<code>Dataset.rdd</code>&nbsp;将&nbsp;<code>Dataset</code>&nbsp;转为&nbsp;<code>RDD</code>&nbsp;的形式</td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
   <td><code>Dataset</code>&nbsp;的执行计划底层的&nbsp;<code>RDD</code></td> 
  </tr> 
 </tbody> 
</table>

可以看到 `(1)` 对比 `(2)` 对了两个步骤, 这两个步骤的本质就是将 `Dataset` 底层的 `InternalRow` 转为 `RDD` 中的对象形式, 这个操作还是会有点重的, 所以慎重使用 `rdd` 属性来转换 `Dataset` 为 `RDD`

总结

1.  `Dataset` 是一个新的 `Spark` 组件, 其底层还是 `RDD`
2.  `Dataset` 提供了访问对象中某个特定字段的能力, 不用像 `RDD` 一样每次都要针对整个对象做操作
3.  `Dataset` 和 `RDD` 不同, 如果想把 `Dataset[T]` 转为 `RDD[T]`, 则需要对 `Dataset` 底层的 `InternalRow` 做转换, 是一个比价重量级的操作

# 2. DataFrame 的作用和常见操作 #

目标

1.  理解 `DataFrame` 是什么
2.  理解 `DataFrame` 的常见操作

`DataFrame` 是什么?

`DataFrame` 是 `SparkSQL` 中一个表示关系型数据库中 `表` 的函数式抽象, 其作用是让 `Spark` 处理大规模结构化数据的时候更加容易. 一般 `DataFrame` 可以处理结构化的数据, 或者是半结构化的数据, 因为这两类数据中都可以获取到 `Schema` 信息. 也就是说 `DataFrame` 中有 `Schema` 信息, 可以像操作表一样操作 `DataFrame`.

![eca0d2e1e2b5ce678161438d87707b61][]

`DataFrame` 由两部分构成, 一是 `row` 的集合, 每个 `row` 对象表示一个行, 二是描述 `DataFrame` 结构的 `Schema`.

![238c241593cd5b0fd06d4d74294680e2][]

`DataFrame` 支持 `SQL` 中常见的操作, 例如: `select`, `filter`, `join`, `group`, `sort`, `join` 等

val spark: SparkSession = new sql.SparkSession.Builder()
      .appName("hello")
      .master("local[6]")
      .getOrCreate()
    
    import spark.implicits._
    
    val peopleDF: DataFrame = Seq(People("zhangsan", 15), People("lisi", 15)).toDF()
    
    /*
    +---+-----+
    |age|count|
    +---+-----+
    | 15|    2|
    +---+-----+
     */
    peopleDF.groupBy('age)
      .count()
      .show()

通过隐式转换创建 `DataFrame`

这种方式本质上是使用 `SparkSession` 中的隐式转换来进行的

val spark: SparkSession = new sql.SparkSession.Builder()
      .appName("hello")
      .master("local[6]")
      .getOrCreate()
    
    // 必须要导入隐式转换
    // 注意: spark 在此处不是包, 而是 SparkSession 对象
    import spark.implicits._
    
    val peopleDF: DataFrame = Seq(People("zhangsan", 15), People("lisi", 15)).toDF()

![841503b4240e7a8ecac62d92203e9943][]

根据源码可以知道, `toDF` 方法可以在 `RDD` 和 `Seq` 中使用

通过集合创建 `DataFrame` 的时候, 集合中不仅可以包含样例类, 也可以只有普通数据类型, 后通过指定列名来创建

val spark: SparkSession = new sql.SparkSession.Builder()
      .appName("hello")
      .master("local[6]")
      .getOrCreate()
    
    import spark.implicits._
    
    val df1: DataFrame = Seq("nihao", "hello").toDF("text")
    
    /*
    +-----+
    | text|
    +-----+
    |nihao|
    |hello|
    +-----+
     */
    df1.show()
    
    val df2: DataFrame = Seq(("a", 1), ("b", 1)).toDF("word", "count")
    
    /*
    +----+-----+
    |word|count|
    +----+-----+
    |   a|    1|
    |   b|    1|
    +----+-----+
     */
    df2.show()

通过外部集合创建 `DataFrame`

不仅可以从 `csv` 文件创建 `DataFrame`, 还可以从 `Table`, `JSON`, `Parquet` 等中创建 `DataFrame`, 后续会有单独的章节来介绍

在 `DataFrame` 上可以使用的常规操作

需求: 查看每个月的统计数量

Step 1: 首先可以打印 `DataFrame` 的 `Schema`, 查看其中所包含的列, 以及列的类型

Step 2: 对于大部分计算来说, 可能不会使用所有的列, 所以可以选择其中某些重要的列

...
    
    df.select('year, 'month, 'PM_Dongsi)

Step 3: 可以针对某些列进行分组, 后对每组数据通过函数做聚合

...
    
    df.select('year, 'month, 'PM_Dongsi)
      .where('PM_Dongsi =!= "Na")
      .groupBy('year, 'month)
      .count()
      .show()

使用 `SQL` 操作 `DataFrame`

使用 `SQL` 来操作某个 `DataFrame` 的话, `SQL` 中必须要有一个 `from` 子句, 所以需要先将 `DataFrame` 注册为一张临时表

val spark: SparkSession = new sql.SparkSession.Builder()
      .appName("hello")
      .master("local[6]")
      .getOrCreate()
    
    val df = spark.read
      .option("header", true)
      .csv("dataset/BeijingPM20100101_20151231.csv")
    
    df.createOrReplaceTempView("temp_table")
    
    spark.sql("select year, month, count(*) from temp_table where PM_Dongsi != 'NA' group by year, month")
      .show()

总结

1.  `DataFrame` 是一个类似于关系型数据库表的函数式组件
2.  `DataFrame` 一般处理结构化数据和半结构化数据
3.  `DataFrame` 具有数据对象的 Schema 信息
4.  可以使用命令式的 `API` 操作 `DataFrame`, 同时也可以使用 `SQL` 操作 `DataFrame`
5.  `DataFrame` 可以由一个已经存在的集合直接创建, 也可以读取外部的数据源来创建

# 3. Dataset 和 DataFrame 的异同 #

目标

1.  理解 `Dataset` 和 `DataFrame` 之间的关系

`DataFrame` 就是 `Dataset`

根据前面的内容, 可以得到如下信息

1.  `Dataset` 中可以使用列来访问数据, `DataFrame` 也可以
2.  `Dataset` 的执行是优化的, `DataFrame` 也是
3.  `Dataset` 具有命令式 `API`, 同时也可以使用 `SQL` 来访问, `DataFrame` 也可以使用这两种不同的方式访问

所以这件事就比较蹊跷了, 两个这么相近的东西为什么会同时出现在 `SparkSQL` 中呢?

![44fb917304a91eab99d131010448331b][]

确实, 这两个组件是同一个东西, `DataFrame` 是 `Dataset` 的一种特殊情况, 也就是说 `DataFrame` 是 `Dataset[Row]` 的别名

`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()

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>DataFrame 就是 Dataset[Row]</td> 
  </tr> 
  <tr> 
   <td>&nbsp;</td> 
   <td>Dataset 的范型可以是任意类型</td> 
  </tr> 
 </tbody> 
</table>

第三点: `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)

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>这行代码明显报错, 无法通过编译</td> 
  </tr> 
 </tbody> 
</table>

`Row` 是什么?

`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)
    }

`DataFrame` 和 `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_fdf: Dataset[People] = df.as[People]
    
    val ds: Dataset[People] = Seq(People("zhangsan", 15), People("lisi", 15)).toDS()
    val df_fds: DataFrame = ds.toDF()

总结

1.  `DataFrame` 就是 `Dataset`, 他们的方式是一样的, 也都支持 `API` 和 `SQL` 两种操作方式
2.  `DataFrame` 只能通过表达式的形式, 或者列的形式来访问数据, 只有 `Dataset` 支持针对于整个对象的操作
3.  `DataFrame` 中的数据表示为 `Row`, 是一个行的概念

[cc610157b92466cac52248a8bf72b76e]: /images/20221022/ebd43fcae6cf46d8a744fa0b6dd45303.png
[eca0d2e1e2b5ce678161438d87707b61]: /images/20221022/f8ddce26b6c24880a8c7ac58b1a744bd.png
[238c241593cd5b0fd06d4d74294680e2]: https://img-blog.csdnimg.cn/img_convert/ce54f1a21acbd01816758c5a7742934f.png
[841503b4240e7a8ecac62d92203e9943]: https://img-blog.csdnimg.cn/img_convert/411ffb26fdaec75dbcc49fb874309f74.png
[44fb917304a91eab99d131010448331b]: https://img-blog.csdnimg.cn/img_convert/fdf14c4d0bec4be373519e662104fd10.png