4、Spark-SQL

1. 学习目标

  • 目标1:掌握Spark SQL原理
  • 目标2:掌握DataFrame/DataSet数据结构和使用方式
  • 目标3:熟练使用Spark SQL完成计算任务

2. Spark SQL

2.1 Spark SQL概述

2.1.1 Spark SQL的历史

  • Shark是一个为Spark设计的大规模数据仓库系统,它与Hive兼容。Shark建立在Hive的代码基础上,并通过将Hive的部分物理执行计划交换出来。这个方法使得Shark的用户可以加速Hive的查询,但是Shark继承了Hive的大且复杂的代码使得Shark很难优化和维护,同时Shark依赖于Spark的版本。随着我们遇到了性能优化的上限,以及集成SQL的一些复杂的分析功能,我们发现Hive的MapReduce设计的框架限制了Shark的发展。在2014年7月1日的Spark Summit上,Databricks宣布终止对Shark的开发,将重点放到Spark SQL上。

2.1.2 什么是Spark SQL

  • Spark SQL是Spark用来处理结构化数据的一个模块,它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用
  • 相比于Spark RDD API,Spark SQL包含了对结构化数据和在其上运算的更多信息,Spark SQL使用这些信息进行了额外的优化,使对结构化数据的操作更加高效和方便。
  • 有多种方式去使用Spark SQL,包括SQL、DataFrames API和Datasets API。但无论是哪种API或者是编程语言,它们都是基于同样的执行引擎,因此你可以在不同的API之间随意切换,它们各有各的特点,看你喜欢那种风格。

2.1.3 Spark SQL的特点

  • Hive,它是将Hive SQL转换成MapReduce然后提交到集群中去执行,大大简化了编写MapReduce程序的复杂性,由于MapReduce这种计算模型执行效率比较慢,所以Spark SQL应运而生,它是将Spark SQL转换成RDD,然后提交到集群中去运行,执行效率非常快!
  • 1.易整合
    • 将sql查询与spark程序无缝混合,可以使用java、scala、python、R等语言的API操作。
  • 2.统一的数据访问
    • 以相同的方式连接到任何数据
  • 3.兼容Hive
    • 支持hiveSQL的语法
  • 4.标准的数据连接
    • 可以使用行业标准的JDBC连接操作关系型数据库

2.2 DataFrame

2.2.1 什么是DataFrame

  • DataFrame的前身是SchemaRDD,从Spark 1.3.0开始SchemaRDD更名为DataFrame。与SchemaRDD的主要区别是:DataFrame不再直接继承自RDD,而是自己实现了RDD的绝大多数功能。你仍旧可以在DataFrame上调用rdd方法将其转换为一个RDD。
  • 在Spark中,DataFrame是一种以RDD为基础的分布式数据集,类似于传统数据库的二维表格,DataFrame带有Schema元信息,即DataFrame所表示的二维表数据集的每一列都带有名称和类型,但底层做了更多的优化。DataFrame可以从很多数据源构建,比如:已经存在的RDD、结构化文件、外部数据库、Hive表。

2.2.2 DataFrame与RDD的区别

  • RDD可看作是分布式的对象的集合,Spark并不知道对象的详细模式信息,DataFrame可看作是分布式的Row对象的集合,其提供了由列组成的详细模式信息,使得Spark SQL可以进行某些形式的执行优化。DataFrame和普通的RDD的逻辑框架区别如下所示:

spark-13

  • 上图直观地体现了DataFrame和RDD的区别。
  • 左侧的RDD[Person]虽然以Person为类型参数,但Spark框架本身不了解 Person类的内部结构。
  • 而右侧的DataFrame却提供了详细的结构信息,使得Spark SQL可以清楚地知道该数据集中包含哪些列,每列的名称和类型各是什么,DataFrame多了数据的结构信息,即schema。这样看起来就像一张表了,DataFrame还配套了新的操作数据的方法,DataFrame API(如df.select())和SQL(select id, name fromxx_table where …)。
  • 此外DataFrame还引入了off-heap,意味着JVM堆以外的内存, 这些内存直接受操作系统管理(而不是JVM)。Spark能够以二进制的形式序列化数据(不包括结构)到off-heap中, 当要操作数据时, 就直接操作off-heap内存. 由于Spark理解schema, 所以知道该如何操作。
  • RDD是分布式的Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外,更重要的特点是提升执行效率、减少数据读取以及执行计划的优化。
  • 有了DataFrame这个高一层的抽象后,我们处理数据更加简单了,甚至可以用SQL来处理数据了,对开发者来说,易用性有了很大的提升。
  • 不仅如此,通过DataFrame API或SQL处理数据,会自动经过Spark 优化器(Catalyst)的优化,即使你写的程序或SQL不高效,也可以运行的很快。

2.2.3 DataFrame与RDD的优缺点

  • RDD的优缺点:
    • 优点:
    • (1)编译时类型安全
      • 编译时就能检查出类型错误
    • (2)面向对象的编程风格
      • 直接通过对象调用方法的形式来操作数据
    • 缺点:
    • (1)序列化和反序列化的性能开销
      • ​无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化。
    • (2)GC的性能开销
      • 频繁的创建和销毁对象,势必会增加GC
  • DataFrame通过引入schema和off-heap(不在堆里面的内存,指的是除了不在堆的内存,使用操作系统上的内存),解决了RDD的缺点, Spark通过schame就能够读懂数据, 因此在通信和IO时就只需要序列化和反序列化数据, 而结构的部分就可以省略了;通过off-heap引入,可以快速的操作数据,避免大量的GC。但是却丢了RDD的优点,DataFrame不是类型安全的, API也不是面向对象风格的。

2.2.4 读取数据源创建DataFrame

读取文本文件创建DataFrame

  • 在spark2.0版本之前,Spark SQL中SQLContext是创建DataFrame和执行SQL的入口,可以利用hiveContext通过hive sql语句操作hive表数据,兼容hive操作,并且hiveContext继承自SQLContext。在spark2.0之后,这些都统一于SparkSession,SparkSession 封装了 SparkContext,SqlContext,通过SparkSession可以获取到SparkConetxt,SqlContext对象。

spark-14

  • (1)在本地创建一个文件,有三列,分别是id、name、age,用空格分隔,然后上传到hdfs上。person.txt内容为:
1
2
3
4
5
6
1 zhangsan 20
2 lisi 29
3 wangwu 25
4 zhaoliu 30
5 tianqi 35
6 kobe 40

上传数据文件到HDFS上:

hdfs dfs -put person.txt /

  • (2)在spark shell执行下面命令,读取数据,将每一行的数据使用列分隔符分割
1
2
先执行 spark-shell --master local[2]
val lineRDD=sc.textFile("/person.txt").map(_.split(" "))

spark-15

  • (3)定义case class(相当于表的schema)

case class Person(id:Int,name:String, age:Int)

spark-16

  • (4)将RDD和case class关联

val personRDD =lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))

spark-17

  • (5)将RDD转换成DataFrame

val personDF = personRDD.toDF

spark-18

  • (6)对DataFrame进行处理

personDF.show

spark-19

personDF.printSchema

spark-20

  • (7)通过SparkSession构建DataFrame

使用spark-shell中已经初始化好的SparkSession对象spark生成DataFrame
val dataFrame=spark.read.text(“/person.txt”)

spark-21

读取json文件创建DataFrame

  • (1)数据文件

    使用spark安装包下的

    /hadoop/apps/server/spark/examples/src/main/resources/people.json文件

  • (2)在spark shell执行下面命令,读取数据

val jsonDF=spark.read.json(“file:///hadoop/apps/server/spark/examples/src/main/resources/people.json”)

spark-22

  • (3)接下来就可以使用DataFrame的函数操作

spark-23

读取parquet列式存储格式文件创建DataFrame

  • (1)数据文件

    使用spark安装包下的

    /hadoop/apps/server/spark/examples/src/main/resources/users.parquet文件

  • (2)在spark shell执行下面命令,读取数据

val parquetDF=spark.read.parquet(“file:///hadoop/apps/server/spark/examples/src/main/resources/users.parquet”)

spark-24

  • (3)接下来就可以使用DataFrame的函数操作

spark-25

2.3 DataFrame常用操作

2.3.1 DSL风格语法

  • DataFrame提供了一个领域特定语言(DSL)来操作结构化数据。

    下面是一些使用示例

  • (1)查看DataFrame中的内容,通过调用show方法

    personDF.show

    spark-26

  • (2)查看DataFrame部分列中的内容

查看name字段的数据
方法一:personDF.select(personDF.col(“name”)).show
方法二:personDF.select(“name”).show

spark-27

查看name和age字段数据

personDF.select(“name”,”age”).show

spark-28

  • (3)打印DataFrame的Schema信息

personDF.printSchema

spark-29

  • (4)查询所有的name和age,并将age+1

personDF.select($"name",$"age",$"age"+1).show

spark-30

  • (5)过滤age大于等于25的,使用filter方法过滤

personDF.filter($”age”>=25).show

spark-31

  • (6)统计年龄大于25的人数

personDF.filter($”age”>=25).count()

spark-32

  • 按年龄进行分组并统计相同年龄的人数

personDF.groupBy(“age”).count().show

spark-33

2.3.2 SQL风格语法

  • DataFrame的一个强大之处就是我们可以将它看作是一个关系型数据表,然后可以通过在程序中使用spark.sql() 来执行SQL语句查询,结果返回一个DataFrame。
  • 如果想使用SQL风格的语法,需要将DataFrame注册成表,采用如下的方式:
1
personDF.registerTempTable("t_person")
  • (1)查询年龄最大的前两名
1
spark.sql("select * from t_person order by age desc limit 2").show

spark-34

  • (2)显示表的Schema信息
1
spark.sql("desc t_person").show

spark-35

  • (3)查询年龄大于25的人的信息
1
spark.sql("select * from t_person where age > 25").show

spark-36

2.4 DataSet

2.4.1 什么是DataSet

  • DataSet是分布式的数据集合,Dataset提供了强类型支持,也是在RDD的每行数据加了类型约束。DataSet是在Spark1.6中添加的新的接口。它集中了RDD的优点(强类型和可以用强大lambda函数)以及使用了Spark SQL优化的执行引擎。DataSet可以通过JVM的对象进行构建,可以用函数式的转换(map/flatmap/filter)进行多种操作。

2.4.2 DataFrame、DataSet、RDD的区别

  • 假设RDD中有两行数据长这样:

spark-37

  • DataFrame中的数据长这样:

spark-38

  • 那么DataSet中的数据长这样:

spark-39

  • DataSet包含了DataFrame的功能,Spark2.0中两者统一,DataFrame表示为DataSet[Row],即DataSet的子集。
    • (1)DataSet可以在编译时检查类型
    • (2)并且是面向对象的编程接口
  • 相比DataFrame,Dataset提供了编译时类型检查,对于分布式程序来讲,提交一次作业太费劲了(要编译、打包、上传运行),到提交到集群运行时才发现错误,这会浪费大量的时间,这也是引入Dataset的一个重要原因。

2.4.3 DataFrame与DataSet互相转换

  • DataFrame和DataSet可以相互转化。
    • (1)DataFrame转为 DataSet
      • df.as[ElementType] 这样可以把DataFrame转化为DataSet。
    • (2)DataSet转为DataFrame
      • ds.toDF() 这样可以把DataSet转化为DataFrame。

2.4.4 创建DataSet

  • (1)通过spark.createDataset创建

spark-40

spark-41

  • (2)通toDS方法生成DataSet

spark-42

  • (3)通过DataFrame转化生成

spark-43

3. 以编程方式执行Spark SQL查询

3.1 编写Spark SQL程序实现RDD转换成DataFrame

  • 前面我们学习了如何在Spark Shell中使用SQL完成查询,现在我们通过IDEA编写Spark SQL查询程序。
  • Spark官网提供了两种方法来实现从RDD转换得到DataFrame,第一种方法是利用反射机制,推导包含某种类型的RDD,通过反射将其转换为指定类型的DataFrame,适用于提前知道RDD的schema。第二种方法通过编程接口与RDD进行交互获取schema,并动态创建DataFrame,在运行时决定列及其类型。
  • 首先在maven项目的pom.xml中添加SparkSQL的依赖。
1
2
3
4
5
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.0.2</version>
</dependency>

3.1.1 通过反射对端Schema

  • Scala支持使用case class类型导入RDD转换为DataFrame,通过case class创建schema,case class的参数名称会被利用反射机制作为列名。这种RDD可以高效的转换为DataFrame并注册为表。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
package com.ctyun.text

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
* RDD转化成DataFrame:利用反射机制
*/
object CaseClassSchema {
def main(args: Array[String]): Unit = {
//1.构建sparkSession指定appName、master地址
val spark: SparkSession = SparkSession.builder().appName("CaseClassSchema").master("local[2]").getOrCreate()
//2.从sparkSession获取sparkContext对象
val sc: SparkContext = spark.sparkContext
//3、加载数据
sc.setLogLevel("WARN")
val data: RDD[String] = sc.textFile("d:\\data\\people.txt")
//4.切分每一行记录
val arrayRDD: RDD[Array[String]] = data.map(_.split(" "))

//5.将RDD与Person类关联
val personRDD: RDD[Person] = arrayRDD.map(x=>Person(x(0).toInt,x(1),x(2).toInt))

//6.创建dataFrame,需要导入隐式转换
import spark.implicits._
val personDF: DataFrame = personRDD.toDF
//todo-------------------DSL语法操作 start--------------
//1、显示DataFrame的数据,默认显示20行
personDF.show()
//2、显示DataFrame的schema信息
personDF.printSchema()
//3、显示DataFrame记录数
println(personDF.count())
//4、显示DataFrame的所有字段
personDF.columns.foreach(x=>print(x + ","))
println()
//5、取出DataFrame的第一行记录
println(personDF.head())
//6、显示DataFrame中name字段的所有值
personDF.select("name").show()
//7、过滤出DataFrame中年龄大于25的记录
personDF.filter($"age">25).show()
//8、统计DataFrame中年龄大于25的人数
println(personDF.filter($"age">25).count())
//9、统计DataFrame中按照年龄进行分组,求每个组的人数
personDF.groupBy($"age").count().show()
//todo-------------------DSL语法操作 end-------------

//todo--------------------SQL操作风格 start-----------
//todo:将DataFrame注册成表
personDF.createTempView("person")
//todo:传入sql语句,进行操作
spark.sql("select * from person").show()
spark.sql("select count(1) from person where age > 25").show()
spark.sql("select age,count(age) from person group by age").show()

//todo--------------------SQL操作风格 end-------------

//关闭
sc.stop()
spark.stop()
}
}

case class Person(id: Int, name: String, age: Int)

3.1.2 通过StructType直接指定Schema

  • 当case class不能提前定义好时,可以通过以下三步创建DataFrame
    • (1)将RDD转为包含Row对象的RDD
    • (2)基于StructType类型创建schema,与第一步创建的RDD相匹配
    • (3)通过sparkSession的createDataFrame方法对第一步的RDD应用schema创建DataFrame
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
package com.ctyun.spark_sql

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

/**
* RDD转换成DataFrame:通过指定schema构建DataFrame
*/
object SparksqlSchema {
def main(args: Array[String]): Unit = {
//1.创建SparkSession指定scheam和master
val spark: SparkSession = SparkSession.builder().appName("SparksqlSchema").master("local[2]").getOrCreate()
//2.通过sparksession创建sparkContext
val sc: SparkContext = spark.sparkContext
sc.setLogLevel("WARN")
//3.获取数据文件
val data: RDD[Array[String]] = sc.textFile("d:\\data\\people.txt").map(_.split(" "))
//4.加载数据到row中
val personRDD: RDD[Row] = data.map(x=>Row(x(0).toInt,x(1),x(2).toInt))
//5.创建schema
val schema: StructType = StructType(Seq(
StructField("id", IntegerType, false),
StructField("name", StringType, false),
StructField("age", IntegerType, false)
))


//利用personRDD与schema创建DataFrame
val personDF: DataFrame = spark.createDataFrame(personRDD,schema)
//8.DSL操作DataFrame数据结果
personDF.show()
personDF.printSchema()
//9.将DataFrame注册成表
personDF.createTempView("person")
//sql语句操作
spark.sql("select * from person").show()

spark.sql("select count(1) from person where age > 25").show()

spark.sql("select age,count(1) from person group by age").show()
//关闭
sc.stop()
spark.stop()

}
}

3.2 编写Spark SQL程序操作HiveContext

  • HiveContext是对应spark-hive这个项目,与hive有部分耦合, 支持hql,是SqlContext的子类,在Spark2.0之后,HiveContext和SqlContext在SparkSession进行了统一,可以通过操作SparkSession来操作HiveContext和SqlContext。

3.2.1 添加pom依赖

1
2
3
4
5
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.11</artifactId>
<version>2.3.0</version>
</dependency>

3.2.2 代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
package com.ctyun.spark_sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
* SparkSQL操作hive sql
*/
object HiveSupport {
def main(args: Array[String]): Unit = {
//1.创建sparksession
val spark: SparkSession = SparkSession.builder()
.appName("HiveSupport")
.master("local[2]")
.config("spark.sql.warehouse.dir","d:\\data\\spark-warehouse")
.enableHiveSupport()//开启支持hive
.getOrCreate()
//2.通过sparksession创建sparkContext
val sc: SparkContext = spark.sparkContext
sc.setLogLevel("WARN")
//3.获取数据文件
val data: RDD[Array[String]] = sc.textFile("d:\\data\\people.txt").map(_.split(" "))
//4.执行hive sql
spark.sql("drop table default.person")
spark.sql("create table if not exists default.person(id int,name string,age int)row format " +
"delimited fields terminated by ' '")

spark.sql("load data local inpath 'data/people.txt' into table default.person")

spark.sql("select * from default.person").show()


//关闭
sc.stop()
spark.stop()

}
}

4. 数据源

4.1 JDBC

  • Spark SQL可以通过JDBC从关系型数据库中读取数据的方式创建DataFrame,通过对DataFrame一系列的计算后,还可以将数据再写回关系型数据库中。

4.1.1 SparkSql从MySQL中加载数据

  • SparkSQL代码:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
package com.ctyun.text

import java.util.Properties

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

//sparksql从mysql中加载数据
object DataFromMysql {
def main(args: Array[String]): Unit = {
//1.创建SparkSession对象
val spark: SparkSession = SparkSession.builder().appName("DataFromMysql").master("local[2]").getOrCreate()
//2.创建properties对象,设置mysql连接
val properties: Properties = new Properties()
val url="jdbc:mysql://localhost:3306/spark"
val url1 = "jdbc:mysql://192.168.137.111:3306/spark"
properties.setProperty("user","root")
properties.setProperty("password","root")
//3.读取mysql中的数据
val personDF: DataFrame = spark.read.jdbc(url1,"student",properties)
//4.显示mysql中表的数据
//personDF.show()
//5.将获取到的mysql中的数据注册到spark中
personDF.createOrReplaceTempView("student_tmp")
val data: DataFrame = spark.sql("select distinct * from student_tmp order by id")
//这里测试目标表数据比较多时 目标表将被清空,数据并未被覆盖
data.write.mode("overwrite").jdbc(url,"student",properties)


//关闭
spark.stop()


}
}