Spark SQL 快速入门系列(3) | DataSet的简单介绍及与DataFrame的交互

不温卜火

发布于 2020-10-28 17:39:21

1.2K0

发布于 2020-10-28 17:39:21

文章被收录于专栏：不温卜火

一. 使用 DataSet 进行编程

DataSet 和 RDD 类似, 但是DataSet没有使用 Java 序列化或者 Kryo序列化, 而是使用一种专门的编码器去序列化对象, 然后在网络上处理或者传输.

虽然编码器和标准序列化都负责将对象转换成字节，但编码器是动态生成的代码，使用的格式允许Spark执行许多操作，如过滤、排序和哈希，而无需将字节反序列化回对象。

DataSet是具有强类型的数据集合，需要提供对应的类型信息。

1.1 创建DataSet

1. 使用样例类的序列得到DataSet

scala> case class Person(name: String, age: Int)
defined class Person
// 为样例类创建一个编码器
scala> val ds = Seq(Person("lisi", 20), Person("zs", 21)).toDS
ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: int]
scala> ds.show
+----+---+
|name|age|
+----+---+
|lisi| 20|
| zs| 21|
+----+---+

2. 使用基本类型的序列得到 DataSet

// 基本类型的编码被自动创建. importing spark.implicits._
scala> val ds = Seq(1,2,3,4,5,6).toDS
ds: org.apache.spark.sql.Dataset[Int] = [value: int]
scala> ds.show
+-----+
|value|
+-----+
| 1|
| 2|
| 3|
| 4|
| 5|
| 6|
+-----+

3. 在实际使用的时候, 很少用到把序列转换成 DataSet, 更多的是通过RDD来得到DataSet

1.2 RDD 和 DataSet 的交互

1. 从 RDD 到 DataSet

使用反射来推断包含特定类型对象的RDD的 schema 。

这种基于反射的方法可以生成更简洁的代码，并且当您在编写Spark应用程序时已经知道模式时，这种方法可以很好地工作。

为 Spark SQL 设计的 Scala API 可以自动的把包含样例类的 RDD 转换成 DataSet.

样例类定义了表结构: 样例类参数名通过反射被读到, 然后成为列名.

样例类可以被嵌套, 也可以包含复杂类型: 像Seq或者Array.

scala> val peopleRDD = sc.textFile("examples/src/main/resources/people.txt")
peopleRDD: org.apache.spark.rdd.RDD[String] = examples/src/main/resources/people.txt MapPartitionsRDD[1] at textFile at <console>:24

scala> case class Person(name: String, age: Long)
defined class Person

scala> peopleRDD.map(line => {val para = line.split(",");Person(para(0),para(1).trim.toInt)}).toDS
res0: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

2. 从 DataSet 到 RDD

调用rdd方法即可
scala> val ds = Seq(Person("lisi", 40), Person("zs", 20)).toDS
ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

// 把 ds 转换成 rdd
scala> val rdd = ds.rdd
rdd: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[8] at rdd at <console>:27

scala> rdd.collect
res5: Array[Person] = Array(Person(lisi,40), Person(zs,20))

二. DataFrame 和 DataSet 之间的交互

1. 从 DataFrame到DataSet

scala> val df = spark.read.json("examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> case class People(name: String, age: Long)
defined class People

// DataFrame 转换成 DataSet
scala> val ds = df.as[People]
ds: org.apache.spark.sql.Dataset[People] = [age: bigint, name: string]

2. 从 DataSet到DataFrame

scala> case class Person(name: String, age: Long)
defined class Person

scala> val ds = Seq(Person("Andy", 32)).toDS()
ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

scala> val df = ds.toDF
df: org.apache.spark.sql.DataFrame = [name: string, age: bigint]

scala> df.show
+----+---+
|name|age|
+----+---+
|Andy| 32|
+----+---+

本次的分享就到这里了

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2020/08/08 ，如有侵权请联系 cloudcommunity@tencent.com 删除

spark