Spark SQL

Myth丶恋晨 2023-09-29 18:51 11阅读 0赞

## 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`进行优化

> [参考][Link 1]

### 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` 等带有表头的文件

--------------------

**三者转换关系图**

case class & .toDF

case class & .toDS

case class & .as case class

.rdd

.toDF

RDD

[Link 1]: http://dblab.xmu.edu.cn/blog/1024-2/