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

2020-10-28 17:39:21 浏览数 (1)

一. 使用 DataSet 进行编程

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

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

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

1.1 创建DataSet

  • 1. 使用样例类的序列得到DataSet
代码语言：javascript复制
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
代码语言：javascript复制
// 基本类型的编码被自动创建. 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.

代码语言：javascript复制
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
代码语言：javascript复制
调用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
代码语言：javascript复制
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
代码语言：javascript复制
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|
 ---- ---

  本次的分享就到这里了

spark python 文件存储

0 人点赞