Fork me on GitHub

Spark原理与实现: Spark SQL编程

字数统计: 3.5k字   |   阅读时长≈ 16分

Spark SQL编程

注意: 建表一定是数据集,对数据集进行建表即用df. 而执行SQL是spark.sql()

SparkSession新的起始点

​ 在老的版本中，SparkSQL提供两种SQL查询起始点：一个叫SQLContext，用于Spark自己提供的SQL查询；一个叫HiveContext，用于连接Hive的查询。
​ 现在是使用的是SparkSession了.实际上就是将两个结合了.所以你很方便的去查寻一个json文件,也可以查询一个hive数据.统一的数据入口
​ SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContex和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext，所以计算实际上是由sparkContext完成的。
在Spark SQL中SparkSession是创建DataFrame和执行SQL的入口

DataFrame和dataset可以相互转换. DataFrame是Dataset里面的一种特殊形式

ResultSet():这个方法很恶心,就是编译期不做类型校验,但是你一运行就会报类型转换异常

DataFrame

创建

1
2
3
4
5
6
在Spark SQL中SparkSession是创建DataFrame和执行SQL的入口，所以DataFrame的创建是来自于sparkSession.从sparkSession中去找.而sparkSession是依赖于sparkContext来构建的.它里面放了一个sparkcontext对象,你可以拿出来. 你也可	以直接通过SparkConf来创建一个SparkSession.但是这种方式它里面也是在构建sparkSession之前构建了SparkContext. 因为SparkContext是用于跟Spark集群连接的.

创建DataFrame有三种方式：
1.通过Spark的数据源进行创建；
2.从一个存在的RDD进行转换；
3.还可以从Hive Table进行查询返回。

1）从Spark数据源进行创建
（1）查看Spark数据源进行创建的文件格式

1
2
scala> spark.read.
csv   format   jdbc   json   load   option   options   orc   parquet   schema   table   text   textFile

（2）读取json文件创建DataFrame

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

（3）展示结果

1
2
3
4
5
6
7
8
scala> df.show   这里面用的比较多的行动算子,我们之前写sparkCore是用collect的这种方式展现的是有结构信息的
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

2）从RDD进行转换
后面讨论
3）从Hive Table进行查询返回
后面讨论

SQL风格语法(主要)

1）创建一个DataFrame(注意在DataSet里面定义的一些函数,我DataFrame也是可以用的)

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

2）对DataFrame创建一个临时表
View(视图),视图和table(表)有什么区别呢?视图使用来查的, 而表是用来增删改查的.因为当前的分布式数据集RDD具有不可变性.

1
scala> df.createOrReplaceTempView("people")  //参数是视图名

3）通过SQL语句实现查询全表

1
2
scala> val sqlDF = spark.sql("SELECT * FROM people")
sqlDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

4）结果展示

1
2
3
4
5
6
7
8
scala> sqlDF.show
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

注意：普通临时表是Session范围内的，如果想应用范围内有效，可以使用全局临时表。使用全局临时表时需要全路径访问，如：global_temp.people

5）对于DataFrame创建一个全局表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
scala> spark.sql("SELECT * FROM global_temp.people").show()
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|

scala> spark.newSession().sql("SELECT * FROM global_temp.people").show()
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

DSL风格语法(次要) (DSL风格叫领域特定语言)

DSL风格叫领域特定语言,就是说,他只能在sparkSQL当中能用,换一个地方就不能用了.很恶心.
DSL语言风格就是使用select,filter,map等这些函数.

1）创建一个DataFrame

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

2）查看DataFrame的Schema信息

1
2
3
4
scala> df.printSchema
root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)

3）只查看”name”列数据

1
2
3
4
5
6
7
8
scala> df.select("name").show()
+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+

4）查看”name”列数据以及”age+1”数据

1
2
3
4
5
6
7
8
scala> df.select($"name", $"age" + 1).show()
+-------+---------+
|   name|(age + 1)|
+-------+---------+
|Michael|     null|
|   Andy|       31|
| Justin|       20|
+-------+---------+

5）查看”age”大于”21”的数据

1
2
3
4
5
6
scala> df.filter($"age" > 21).show()
+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+

6）按照”age”分组，查看数据条数

1
2
3
4
5
6
7
8
9
10
scala> df.groupBy("age").count().show()
+----+-----+
| age|count|
+----+-----+
|  19|     1|
|null|     1|
|  30|     1|
+----+-----+


创建DataFrame有三种方式：

1
2
3
4.通过Spark的数据源进行创建；
5.从一个存在的RDD进行转换；
6.还可以从Hive Table进行查询返回。

RDD转换为DataFrame

注意：如果需要RDD与DF(DataFrame)或者DS(DataSet)之间操作，那么都需要引入 import spark.implicits._ （spark不是包名，而是sparkSession对象的名称）

前置条件：导入隐式转换并创建一个RDD

1
2
3
4
5
scala> import spark.implicits._
import spark.implicits._

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[3] at textFile at <console>:27

1）通过手动确定转换

1
2
scala> peopleRDD.map{x=>val para = x.split(",");(para(0),para(1).trim.toInt)}.toDF("name","age")
res1: org.apache.spark.sql.DataFrame = [name: string, age: int]

2）通过反射确定（需要用到样例类）
（1）创建一个样例类,样例类其实就是和Java中的类一样

1
scala> case class People(name:String, age:Int)

（2）根据样例类将RDD转换为DataFrame

1
2
scala> peopleRDD.map{ x => val para = x.split(",");People(para(0),para(1).trim.toInt)}.toDF
res2: org.apache.spark.sql.DataFrame = [name: string, age: int]

3）通过编程的方式（了解）
（1）导入所需的类型

1
2
scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._

（2）创建Schema

1
2
scala> val structType: StructType = StructType(StructField("name", StringType) :: StructField("age", IntegerType) :: Nil)
structType: org.apache.spark.sql.types.StructType = StructType(StructField(name,StringType,true), StructField(age,IntegerType,true))

（3）导入所需的类型

1
2
scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row

（4）根据给定的类型创建二元组RDD

1
2
scala> val data = peopleRDD.map{ x => val para = x.split(",");Row(para(0),para(1).trim.toInt)}
data: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[6] at map at <console>:33

（5）根据数据及给定的schema创建DataFrame

1
2
scala> val dataFrame = spark.createDataFrame(data, structType)
dataFrame: org.apache.spark.sql.DataFrame = [name: string, age: int]

DataFrame转换为RDD

直接调用rdd即可
1）创建一个DataFrame

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

2）将DataFrame转换为RDD

1
2
scala> val dfToRDD = df.rdd
dfToRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[19] at rdd at <console>:29

3）打印RDD

1
2
scala> dfToRDD.collect
res13: Array[org.apache.spark.sql.Row] = Array([Michael, 29], [Andy, 30], [Justin, 19])

DataSet

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

创建

1）创建一个样例类

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

2）创建DataSet

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

RDD转换为DataSet

SparkSQL能够自动将包含有case类的RDD转换成DataFrame，case类定义了table的结构，case类属性通过反射变成了表的列名。Case类可以包含诸如Seqs或者Array等复杂的结构。
1）创建一个RDD

1
2
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[3] at textFile at <console>:27

2）创建一个样例类

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

3）将RDD转化为DataSet

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

DataSet转换为RDD

调用rdd方法即可。
1）创建一个DataSet

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

2）将DataSet转换为RDD

1
2
scala> DS.rdd
res11: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[15] at rdd at <console>:28

DataFrame与DataSet的互操作

DataFrame转DataSet

1）创建一个DateFrame

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

2）创建一个样例类

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

3）将DataFrame转化为DataSet

1
2
scala> df.as[Person]
res14: org.apache.spark.sql.Dataset[Person] = [age: bigint, name: string]

Dataset转DataFrame

1）创建一个样例类

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

2）创建DataSet

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

3）将DataSet转化为DataFrame

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

4）展示

1
2
3
4
5
6
scala> df.show
+----+---+
|name|age|
+----+---+
|Andy| 32|
+----+---+

这种方法就是在给出每一列的类型后，使用as方法，转成Dataset，这在数据类型是DataFrame又需要针对各个字段处理时极为方便。在使用一些特殊的操作时，一定要加上 import spark.implicits._ 不然toDF、toDS无法使用。

RDD、DataFrame和DataSet

在SparkSQL中Spark为我们提供了两个新的抽象，分别是DataFrame和DataSet。他们和RDD有什么区别呢？首先从版本的产生上来看:

1
2
RDD (Spark1.0) —> Dataframe(Spark1.3) —> Dataset(Spark1.6)
如果同样的数据都给到这三个数据结构，他们分别计算之后，都会给出相同的结果。不同是的他们的执行效率和执行方式。在后期的Spark版本中，DataSet有可能会逐步取代RDD和DataFrame成为唯一的API接口。

三者的共性

1
2
3
4
（1）RDD、DataFrame、DataSet全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利;
（2）三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算;
（3）三者有许多共同的函数，如filter，排序等;
（4）在对DataFrame和Dataset进行操作许多操作都需要这个包:import spark.implicits._（在创建好SparkSession对象后尽量直接导入）,DataFrame和DataSet特别是和RDD进行操作的时候,则需要引入一个隐式转换.   因为这个spark session对象是由我们启动的spark-shell来给我们创建的,这个spark-shell窗口在启动的时候就已经将隐式转换导入进来了. 那就算把RDD在代码里面转成了样例类对象了,你调用.toDS调用不出来. 因为你没有加隐式转换.  而且这个隐式转换mport spark.implicits._  这个spark看起来像包,但是他不是, 这个spark是sparkSession的对象的对象

三者的互相转化

1
2
3
4
5
6
7
8
RDD到DataFrame或者到dataSet,以及DataFrame到DataSet,这个过程我们认为是由简单到复杂,需要加东西的过程. 这一套过程都需要用到一个样例类. 
RDD到DataFrame用的是.toDF
RDD到DataSet用的是.toDS
DataFrame到DataSet用的是.as[样例类]

DataFrame和dataSet到RDD或者Dataset到DataFrame 这是简的过程
DataFrame和DataSet到RDD直接使用.rdd 即可
DataSet到DataFrame使用.toDF

IDEA创建SparkSQL程序

1）添加依赖

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

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
40
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

object HelloWorld {

  def main(args: Array[String]) {

    //创建SparkConf()并设置App名称
    val spark = SparkSession
      .builder()

.master("local[*]")
      .appName("HelloWorld")
      //.config("spark.some.config.option", "some-value")
      .getOrCreate()

//导入隐式转换
import spark.implicits._

//读取本地文件，创建DataFrame
val df = spark.read.json("examples/src/main/resources/people.json")

//打印
df.show()

//DSL风格：查询年龄在21岁以上的
df.filter($"age" > 21).show()

//创建临时表
df.createOrReplaceTempView("persons")

//SQL风格：查询年龄在21岁以上的
spark.sql("SELECT * FROM persons where age > 21").show()

//关闭连接
spark.stop()

}

}

用户自定义函数

在Shell窗口中可以通过spark.udf功能用户可以自定义函数。

UDF

1）创建DataFrame

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

2）打印数据

1
2
3
4
5
6
7
8
scala> df.show()
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

3）注册UDF，功能为在数据前添加字符串
因为自定义函数,最终是在SQL里面去用,所以你得有函数名addName, 后面要有函数如何添加? 相当于你在hive当中自定义的函数. 

1
2
scala> spark.udf.register("addName", (x:String)=> "Name:"+x)
res5: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

4）创建临时表

1
scala> df.createOrReplaceTempView("people")

5）应用UDF

1
2
3
4
5
6
7
8
scala> spark.sql("Select addName(name), age from people").show()
+-----------------+----+
|UDF:addName(name)| age|
+-----------------+----+
|     Name:Michael|null|
|        Name:Andy|  30|
|      Name:Justin|  19|
+-----------------+----+

UDAF

1
2
3
4
5
6
7
自定义UDAF函数,多进一出,聚合函数.
多进:关心进来的数据是什么样子的
出:出去看你做什么事情
对于Spark程序和MR程序,他是一行一行读取的.


强类型的Dataset和弱类型的DataFrame都提供了相关的聚合函数， 如 count()，countDistinct()，avg()，max()，min()。除此之外，用户可以设定自己的自定义聚合函数。通过继承UserDefinedAggregateFunction来实现用户自定义聚合函数。

1）需求：实现求平均工资的自定义聚合函数。
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
40
41
42
43
44
45
46
47
48
49
import org.apache.spark.sql.expressions.MutableAggregationBuffer
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession

object MyAverage extends UserDefinedAggregateFunction {

// 聚合函数输入参数的数据类型
def inputSchema: StructType = StructType(StructField("inputColumn", LongType) :: Nil)

// 聚合缓冲区中值得数据类型
def bufferSchema: StructType = {
StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)
}

// 返回值的数据类型
def dataType: DataType = DoubleType

// 对于相同的输入是否一直返回相同的输出。
def deterministic: Boolean = true

// 初始化
def initialize(buffer: MutableAggregationBuffer): Unit = {

// 存工资的总额
buffer(0) = 0L

// 存工资的个数
buffer(1) = 0L
}

// 同一个分区数据合并
def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
if (!input.isNullAt(0)) {
buffer(0) = buffer.getLong(0) + input.getLong(0)
buffer(1) = buffer.getLong(1) + 1
}
}

// 不同分区间数据合并
def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
}

// 计算最终结果
def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)
}

3）函数使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 注册函数
spark.udf.register("myAverage", MyAverage)

val df = spark.read.json("examples/src/main/resources/employees.json")
df.createOrReplaceTempView("employees")
df.show()
// +-------+------+
// |   name|salary|
/ +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

val result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees")
result.show()
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0 |
// +--------------+
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

标签列表:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
大数据开发、Java后端开发
联系QQ:3101001329
微信:shangbaishuyao

很惭愧

只做了一点微小的工作
谢谢大家