1. Spark SQL 概述

Spark SQL 是 spark 用来处理结构化数据的模块，它提供了2个编程抽象, 类似 Spark Core 中的 RDD：

• DataFrame
• DataSet

1.1 DataFrame 与 RDD 的区别

DataFrame 是一个分布式数据容器，类似于一张二维表，与 rdd 相比，它还存储了数据的数据类型，数据的结构信息，即 schema，可以清楚地知道该数据集中包含哪些列、每列的名称和类型。

另外 DataFrame 的各种变换操作也采用惰性机制，只是记录了各种转换的逻辑转换路线图（是一个DAG图），不会发生真正的计算，这个DAG图相当于一个逻辑查询计划，最终，会被翻译成物理查询计划，生成RDD DAG。

性能上要比 RDD 要高，主要因为 DataFrame 优化的执行计划：查询计划通过 Spark catalyst optimiser进行优化

参考

1.2 DataSet

• 是 DataFrame API 的一个扩展，是 SparkSQL 最新的数据抽象(1.6新增)
• 用户友好的 API 风格，既具有类型安全检查也具有 DataFrame 的查询优化特性
• Dataset 支持编解码器，当需要访问非堆上的数据时可以避免反序列化整个对象，提高了效率
• 样例类被用来在 DataSet 中定义数据的结构信息，样例类中每个属性的名称直接映射到 DataSet 中的字段名称
• DataFrame 是 DataSet 的特列，DataFrame=DataSet[Row] ，所以可以通过as方法将DataFrame转换为DataSet。Row是一个类型，跟Car、Person这些的类型一样，所有的表结构信息都用Row来表示
• DataSet 是强类型的。比如可以有DataSet[Car]，DataSet[Person]
• DataFrame只是知道字段，但是不知道字段的类型，所以在执行这些操作的时候是没办法在编译的时候检查是否类型失败的，比如你可以对一个String进行减法操作，在执行的时候才报错，而DataSet不仅仅知道字段，而且知道字段类型，所以有更严格的错误检查。就跟JSON对象和类对象之间的类比

2. 创建 DataFrame

从2.0开始, SparkSession是 Spark 最新的 SQL 查询起始点，实质上是SQLContext和HiveContext的组合，其内部封装了SparkContext，所以计算实际上是由SparkContext完成的，DataFrame 的创建需要用到 SparkSession

有三种方式可以来创建 DataFrame:

• 通过 Spark 的数据源创建
• 已存在的 rdd 转换而来
• 读取 hive 数据转换

SparkSession 的使用方法：

import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
val sc = spark.sparkContext

2.1 通过 Spark 的数据源创建

• jdbc：重点
• json
• parquet：根据需求选择
• hive：重点
• scala 集合

// 读取本地 spark 提供的 json 数据源
scala> val df = spark.read.json("file:///home/hadoop/apps/spark-2.2.0/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> df.show
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

2.2 从 RDD 转换得到 DataFrame

RDD 转换得到 DataFrame 有两种方式：

• 利用反射来推断包含特定类型对象的 RDD 的 schema
• 使用编程接口，构造一个 schema 并将其应用在已知的 RDD 上

2.2.1 利用反射机制推断 RDD 模式

反射机制模式需要事先定义一个样例类 ：case，只有样例类才能被转换 （需要导入 import spark.implicits._，否则不能使用 toDF()）：

import org.apache.spark.sql.SparkSession

// 样例类
case class People(name: String, age: Int)

object CreateDF {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
    val sc = spark.sparkContext

    // 隐士转换，否则用不了 toDF，支持把一个RDD隐式转换为一个DataFrame
    import spark.implicits._

    val rdd = sc.textFile("hdfs://hadoop1:9000/people.txt").
      map(_.split(",")).
      map(x => People(x(0), x(1).trim.toInt))

    val df = rdd.toDF()
    df.show()
    rdd.collect().foreach(println)
    sc.stop()
    spark.stop()
  }

}

运行结果：

+-------+---+
|   name|age|
+-------+---+
|Michael| 29|
|   Andy| 30|
| Justin| 19|
+-------+---+

People(Michael,29)
People(Andy,30)
People(Justin,19)

2.2.2 使用编程方式定义RDD模式

当无法定义样例类时，可以采用此种方式：

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object CreateDF {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
    val sc = spark.sparkContext
	
      // rdd 必须是 Row 类型
    val rdd = sc.textFile("hdfs://hadoop1:9000/people.txt").
      map(_.split(",")).
      map(x => Row(x(0), x(1).trim.toInt))

    val schema = StructType(Array(
      StructField("name", StringType, true),
      StructField("age", IntegerType, true)
    ))

    val df = spark.createDataFrame(rdd, schema)

    df.show()
    rdd.collect().foreach(println)
    sc.stop()
    spark.stop()
  }

}

注意：rdd 必须 Row 类型！

3. DataFrame 语法风格

DataFrame 主要有两种语法风格：

• SQL ：主要，查询数据的时候使用 SQL 语句来查询，必须要有临时视图或者全局视图来辅助
• DSL：一种特定的语言风格，用于管理结构化的数据. 可以在 Scala, Java, Python 和 R 中使用 DSL，不需要创建临时视图

3.1 SQL 风格

视图主要分为三种：

• createTempView：创建一个临时视图，当视图名称存在时会报错，不推荐
• createOrReplaceTempView：创建一个临时视图，当视图名称存在时会选择替换，推荐，只在当前 session 有效
• createOrReplaceGlobalTempView：创建一个全局临时视图，在所有 session 都有效

使用示例：

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

scala> df.create
createGlobalTempView   createOrReplaceGlobalTempView   createOrReplaceTempView   createTempView

scala> df.createOrReplaceTempView("people")

scala> spark.sql("select * from people")
res2: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> spark.sql("select * from people").show
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+		

// 全局临时视图，访问格式：global_temp  + 视图名称

scala> df.createOrReplaceGlobalTempView("p1")

scala> spark.sql("select * from global_temp.p1").show

3.2 DSL 风格

DSL 风格不需要创建临时视图，但是涉及到运算操作（如：对列元素进行加减乘除、逻辑判断等）都需要在列名前面加上 $，如：$name：

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

scala> df.select("name").show
+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+


scala> df.select("name", "age").show
+-------+----+
|   name| age|
+-------+----+
|Michael|null|
|   Andy|  30|
| Justin|  19|
+-------+----+

scala> df.select($"name", $"age" + 1).show
+-------+---------+
|   name|(age + 1)|
+-------+---------+
|Michael|     null|
|   Andy|       31|
| Justin|       20|
+-------+---------+


scala> df.select($"name", $"age" > 20).show
+-------+----------+
|   name|(age > 20)|
+-------+----------+
|Michael|      null|
|   Andy|      true|
| Justin|     false|
+-------+----------+

// 分组查询

scala> df.groupBy("age").count.show
+----+-----+
| age|count|
+----+-----+
|  19|    1|
|null|    1|
|  30|    1|
+----+-----+

4. RDD 和 DataFrame 交互

RDD、DF、DS 三者之间可以相互转换，另外需要导入：import spark.implicits._，这里的 spark 指的是 SparkSession 的那个对象，所以需要先创建 SparkSession 再导入 implicits

4.1 RDD 转换 DF

三种方式：

• toDF() 手动转换

• 通过样例类反射转换（最常用）：详情见 2.2.1 利用反射机制推断 RDD 模式

• 通过 API 的方式转换（不推荐）：详情见 2.2.2 使用编码方式定位 RDD 模式

手动转换

scala>  val rdd = sc.textFile("hdfs://hadoop1:9000/people.txt")
rdd: org.apache.spark.rdd.RDD[String] = hdfs://hadoop1:9000/people.txt MapPartitionsRDD[8] at textFile at <console>:24

scala> val rdd2 = rdd.map(line => {val paras = line.split(","); (paras(0), paras(1).trim.toInt)})
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[9] at map at <console>:26

scala> rdd2.collect().foreach(print)
(Michael,29)(Andy,30)(Justin,19)

scala> rdd2.toDF("name", "age").show
+-------+---+
|   name|age|
+-------+---+
|Michael| 29|
|   Andy| 30|
| Justin| 19|
+-------+---+

4.2 DF 转 RDD

DF 转 RDD 只需调用 .rdd 即可，得到是一个 Row 类型：

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

scala> df.rdd
res4: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[20] at rdd at <console>:26

scala> df.rdd.collect()
res6: Array[org.apache.spark.sql.Row] = Array([null,Michael], [30,Andy], [19,Justin])

scala> df.rdd.collect()(0)
res7: org.apache.spark.sql.Row = [null,Michael]

scala> df.rdd.collect()(0)(0)
res8: Any = null

scala> df.rdd.collect()(0)(1)
res9: Any = Michael

5. DataSet

DS 与 RDD 类似，但是没有使用 Java/Kryo 序列化，而是采用一种专门的编码器序列化对象，然后在网络上处理或传输。虽然编码器和标准序列化都负责将对象转换成字节，但编码器是动态生成的代码，使用的格式允许Spark执行许多操作，如过滤、排序和哈希，而无需将字节反序列化回对象。

另外 DS 是强类型的数据集合，需要提供对应的类型信息，它是在 DF 的基础上进行了额外的拓展，同时又拥有 DF 所具备的功能。

5.1 创建 DS

1、基本类型的序列创建：

scala> import spark.implicits._
import spark.implicits._

scala> val list1 = List(1, 2, 3, 4, 5, 6)
list1: List[Int] = List(1, 2, 3, 4, 5, 6)

scala> val ds = list1.toDS()
ds: org.apache.spark.sql.Dataset[Int] = [value: int]

// 未指定列名，默认为 value
scala> ds.show
+-----+
|value|
+-----+
|    1|
|    2|
|    3|
|    4|
|    5|
|    6|
+-----+

2、使用样例类创建：

import org.apache.spark.sql.{Dataset, SparkSession}

case class Info(name: String, age: Int)

object CreateDS {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
    val sc = spark.sparkContext
    import spark.implicits._

    // 方法一：基本类型的序列创建
    val list1 = List(1, 2, 3, 4, 5, 6)
    val ds: Dataset[Int] = list1.toDS()

//    // 方法二：使用样例类创建
//    val list = List(Info("rose", 18), Info("lila", 19), Info("john", 20))
//    val ds = list.toDS()

    ds.show()

    sc.stop()
    spark.stop()
  }
}

注意：两种方式都要导入 import spark.implicits._，实际使用中更多的是通过 rdd 转换为 ds

5.2 RDD 与 DS 交互

5.2.1 RDD 转换为 DS

通过样例类转换：

import org.apache.spark.sql.SparkSession

case class User(name: String, age: Int)

object RDD2DS {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
    val sc = spark.sparkContext
    import spark.implicits._

    val rdd = sc.textFile("hdfs://hadoop1:9000/people.txt").
      map(_.split(",")).
      map(x => User(x(0), x(1).trim.toInt))

    val ds = rdd.toDS()
    ds.show()

    sc.stop()
    spark.stop()
  }
}

5.2.2 DS 转换为 RDD

scala> ds.rdd
res12: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[24] at rdd at <console>:31

5.3 DF 与 DS 交互

5.3.1 DF 转换为 DS

package top.midworld.spark1031.create_df

import org.apache.spark.sql.SparkSession

case class PeopleTest(name: String, age: Long)

object DF2DS {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
    val sc = spark.sparkContext
    import spark.implicits._

    val df = spark.read.json("hdfs://hadoop1:9000/people.json")

    val ds = df.as[PeopleTest]
    ds.show()
    sc.stop()
    spark.stop()
  }
}

5.3.2 DS 转换为 DF

import org.apache.spark.sql.SparkSession

case class PeopleTest(name: String, age: Long)

object DS2DF {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("create_rdd").master("local[2]").getOrCreate()
    val sc = spark.sparkContext
    import spark.implicits._

    val list = List(PeopleTest("rose", 18), PeopleTest("lila", 19), PeopleTest("john", 20))
    val ds = list.toDS()
    val df = ds.toDF()

    df.show()
    sc.stop()
    spark.stop()
  }
}

6. RDD 、DF和 DS 的关系

RDD (Spark1.0) —> Dataframe(Spark1.3) —> Dataset(Spark1.6)

共同点

• 三者都是惰性的，只有在执行 Action 才会触发计算
• 三者会根据 spark 内存情况自动缓存运算
• 都有 partition 分区概念
• 在对 DataFrame 和 Dataset 进行操作许多操作都需要这个包进行支持 import spark.implicits._
• DF、DS 均可使用模式匹配获取各个字段的值和类型

区别

• rdd 一般与 spark mlib 机器学习同时使用
• df 每行类型固定为 Row，每列的值无法直接访问，只有解析后才能获取；ds 每行的数据类型不是固定的，可以根据样例类获得每行的信息
• df、ds 都可以使用 spark sql 操作，都支持 sql、dsl 两种风格
• df、ds 也可以保存为 Excel、csv 等带有表头的文件

三者转换关系图