大数据学习之SparkSql

95.SPARKSQL_简介

网址： https://spark.apache.org/sql/

Spark SQL 是 Spark 的一个模块，用于处理 结构化的数据 。

SparkSQL 特点

1

易整合

无缝的整合了 SQL 查询和 Spark 编程，随时用 SQL 或

DataFrame API 处理结构化数据。并且支持多语言 Java 、

Scala 、 Python 、 R 。

2

统一的数据访问

使用相同的方式连接不同的数据源或不同的文件格式中的数

据。

3

兼容 Hive

在已有的数据仓库上直接运行 SQL 或者 HiveQL ，也可以使

用 SparkSQL 直接处理数据并生成 Hive 数据表。

1 4

标准的数据连接

支持标准化的 JDBC\ODBC 连接 , 方便和各种数据进行数据交换

实时效果反馈

1. 关于 SparkSQL 特点的描述，错误的是：

A

易整合：无缝的整合了 SQL 查询和 Spark 编程，随时用

SQL 或 DataFrame API 处理结构化数据。并且支持多语言 Java 、

Scala 、 Python 、 R 。

B

统一的数据访问：使用相同的方式连接不同的数据源或不同

的文件格式中的数据。

C

不兼容 Hive 。

D

支持标准化的 JDBC\ODBC 连接 , 方便和各种数据进行数据交

换。

答案：

1=>C 兼容 Hive

96.SPARKSQL_发展史

Hive 是早期唯一运行在 Hadoop 上的 SQL-on-Hadoop 工具。但

是 MapReduce 计算过程中大量的中间磁盘落地过程消耗了大量的

I/O ，降低的运行效率，为了提高 SQL-on-Hadoop 的效率，大量的

SQL-on-Hadoop 工具开始产生，其中表现较为突出的是： Impala 、

Shark 、 Drill 。

其中 Shark 是伯克利实验室 Spark 生态环境的组件之一，是基于

Hive 所开发的工具。 Shark 对于 Hive 的太多依赖，制约了 Spark 各个

组件的相互集成，所以提出了 SparkSQL 项目。

SparkSQL 抛弃原有 Shark 的代码，汲取了 Shark 的一些优点，

如内存列存储（ In-Memory Columnar Storage ）、 Hive 兼容性

等，重新开发了 SparkSQL 代码；由于摆脱了对 Hive 的依赖性，

SparkSQL 无论在数据兼容、性能优化、组件扩展方面都得到了极大

的方便。

相关时间节点

1

2014 年 1.0 正式发布， Shark 项目和 SparkSQL 项目的主持人

Reynold Xin 宣布：停止对 Shark 的开发，团队将所有资源放

SparkSQL 项目上，至此， Shark 的发展画上了句话，但也因

此发展出两个支线： SparkSQL 和 Hive on Spark 。

3 2

2015 年 1.3 发布 DataFrame 数据结构 , 沿用至今

3

2016 年 1.6 发布 Dataset 数据结构 ( 带泛型的 DataFrame), 适

用于支持泛型的语言 Java 、 Scala

4

2016 年 2.0 统一了 Dataset 和 DataFrame, 以后只有 Dataset

了 , Python 用的 DataFrame 就是没有泛型的 Dataset 。

5

2019 年 3.0 发布， 性能大幅度提升， SparkSQL 变化不大。

6

2021 年 3.2 发布，支持 Java 、 Python 、 Scala 、 R

总结：

SparkSQL 用于处理大规模结构化数据的计算引擎

1

SparkSQL 在企业中广泛使用，并性能极好。

2

SparkSQL ：使用简单、 API 统一、兼容 HIVE 、支持标准化 JDBC 和 ODBC 连接

3

SparkSQL 2014 年正式发布，当下使用最多的 2.3 版

4

Spark 发布于 2016 年，当下使用的最新 3.2.1 办发布于 2022 年

97.SPARKSQL_与HIVE区别

98.SPARKSQL_SPARKSESSION

99.SPARKSQL_数据抽象

100.SPARKSQL_DATAFRAME概述

101.SPARKSQL_DATASET概述

102.SPARKSQL_DATAFRAME构成

103.SPARKSQL_创建项目

104.SPARKSQL_createDataFrame创建DF

package com . itbaizhan . sql

import org . apache . spark . SparkConf

import org . apache . spark . rdd . RDD

import org . apache . spark . sql . types .

{ IntegerType , StringType , StructField ,

StructType }

import org . apache . spark . sql .{ DataFrame , Row ,

SparkSession }

1

2

3

4

5

6

7

15 object CreateDataFrame {

def main ( args : Array [ String ]): Unit = {

//1. 创建上下文配置文件对象

val conf = new

SparkConf (). setMaster ( "local[*]" )

    . setAppName ( "CreateDataFrame" )

//2. 创建执行环境入口 SparkSession 对象

val spark : SparkSession =

SparkSession . builder ()

    . config ( conf ). getOrCreate ()

//3. 读取文件，映射创建 RDD[Row] 对象

val rdd : RDD [ Row ] = spark . sparkContext

    . textFile ( "data/sql/student.txt" )

    . map ( _ . split ( "," ))

    . map ( array => Row ( array ( 0 ). toInt ,

array ( 1 ). trim , array ( 2 ). toInt ))

//4. 定义 StructType 对象，指定所有列名和各自的类

型

val schema : StructType = StructType (

StructField ( "id" , IntegerType , false )

::

StructField ( "name" , StringType ,

false ) ::

StructField ( "age" , IntegerType ,

true ) :: Nil )

//5. 基于 rdd 对象转为 DataFrame

val df : DataFrame =

spark . createDataFrame ( rdd , schema )

//6. 打印 df 的表结构信息

df . printSchema ()

//7. 输出 df 中的数据

df . show ()

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

16 运行输出：

SparkSQL_toDF创建DF

将 RDD 转换为 DataFrame ：使用 RDD 的 toDF 方法转换 RDD

// 关闭

spark . stop ()

}

}

32

33

34

35

# 打印 DataFrame 的表结构

root

|-- id: long (nullable = true)

|-- name: string (nullable = true)

|-- age: long (nullable = true)

#df 中的数据

+---+-----+---+

|id |name |age|

+---+-----+---+

|1 |tom |22 |

|2 |lucy |21 |

|3 |peter|23 |

|4 |john |25 |

+---+-----+---+

105.SPARKSQL_toDF创建DF

在 IDEA 中开发程序时，如果需要 RDD 与 DF 或者 DS 之间互相转

换操作，那么需要引入

import spark.implicits._

package com . itbaizhan . sql

import org . apache . spark . SparkConf

import org . apache . spark . rdd . RDD

import org . apache . spark . sql .{ DataFrame ,

SparkSession }

object ToDFDemo1 {

def main ( args : Array [ String ]): Unit = {

//1. 创建配置文件对象

val conf : SparkConf = new SparkConf ()

. setMaster ( "local[*]" ). setAppName ( "toDF" )

//2. 创建 SparkSession 对象

val spark : SparkSession =

SparkSession . builder (). config ( conf ). getOrCre

ate ()

//4. 添加隐式转换

import spark . implicits . _

//5. 读取本地文件，并映射创建 RDD

val rdd : RDD [( Int , String , Int )] =

spark . sparkContext

    . textFile ( "data/sql/student.txt" )

//RDD[String]"1,tom,22"-

>RDD[Array[String]]

    . map ( _ . split ( "," ))

//RDD[Array[String]]->RDD[(Int,

String, Int)]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

18 运行输出：

    . map ( arr => ( arr ( 0 ). toInt ,

arr ( 1 ). trim , arr ( 2 ). toInt ))

//6. 通过 rdd.toDF(colNames: String*)

//val df: DataFrame = rdd.toDF()// 了解

val df : DataFrame = rdd . toDF ( "id" ,

"name" , "age" )

//7. 输出结构信息

df . printSchema ()

//8. 输出 df 中的数据

/**show(numRows: Int, truncate: Boolean)

* numRows: 表示输出数据的行数 , 默认是 20 行 .

* truncate: 表示输出时是否对列的值进行截取

*     false: 表示不截取

*     true: 表示截取，保留 20 个字符

*/

//df.show()

//df.show(2,false)

df . show ( 10 , false )

//3. 关闭 spark

spark . stop ()

}

}

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

19 SparkSQL_toDF 使用样例类创建 DF

实际开发中，一般通过样例类将 RDD 转换为 DataFrame

在 IDEA 中开发程序时，如果需要 RDD 与 DF 或者 DS 之间互相转

换操作，那么需要引入

import spark.implicits._

定义样例类 Student ：

root

|-- id: integer (nullable = false)

|-- name: string (nullable = true)

|-- age: integer (nullable = false)

+---+-----+---+

| id| name|age|

+---+-----+---+

| 1| tom| 22|

| 2| lucy| 21|

| 3|peter| 23|

| 4| john| 25|

+---+-----+---+

106.SPARKSQL_toDF使用样例创建DF

107.SPARKSQL_DataFrame转换RDD

108.SPARKSQL_创建DATASET

109.SPARKSQL_DATASET和RDD转换

110.SPARKSQL_DATASET和DATAFRAME转换

111.SPARKSQL_读写parquet文件

https://spark.apache.org/docs/latest/sql-data-sources.html

112.SPARKSQL_读写parquet文件扩展

113.SPARKSQL_读写text文件

114.SPARKSQL_读写JSON文件

115.SPARKSQL_读写csv文件

116.SPARKSQL_JDBC读写MYSQL

117.SPARKSQL_JDBC写MYSQL

118.SPARKSQL_SPARKONHIVE概述

119.SPARKSQL_SPARKONHIVE配置

120.SPARKSQL_SPARKONHIVE代码开发一

121.SPARKSQL_SPARKONHIVE代码开发二

122.SPARKSQL_SPARKONHIVE代码集群提交

123.SPARKSQL_DSL_API

124.SPARKSQL_数据去重

125.SPARKSQL_function包

126.SPARKSQL_SQL API

127.SPARKSQL_SQL API实战

128.SPARKSQL_自定义函数概述

129.SPARKSQL_自定义UDF函数

130.SPARKSQL_自定义UDF函数扩展

131.SPARKSQL_ARRAYTYPE返回值类型的UDF

132.SPARKSQL_UDAF函数OLD一

133.SPARKSQL_UDAF函数OLD二

134.SPARKSQL_UDAF函数OLD三

135.SPARKSQL_UDAF函数OLD四

136.SPARKSQL_UDAF函数NEW一

137.SPARKSQL_UDAF函数NEW二

138.SPARKSQL_UDAF函数NEW三

139.SPARKSQL_开窗函数概述

140.SPARKSQL_开窗函数实战

141.SPARKSQL实战_找出变化的行一

142.SPARKSQL实战_找出变化的行二

143.SPARKSQL实战_函数转换JSON数据

144.SPARKSQL实战_读取嵌套的json

145.SPARKSQL实战_解析JSONARRAY数据

146.SPARKSQL实战_行转换行一

147.SPARKSQL实战_行转换行二

148.SPARKSQL实战_行转换行三

149.SPARKSQL实战_行转换行四

150.SPARKSQL实战_用户7日留存分析一

151.SPARKSQL实战_用户7日留存分析二

152.SPARKSQL实战_用户7日留存分析三

153.SPARKSQL实战_统计访问总时长一

154.SPARKSQL实战_统计访问总时长二

155.SPARKSQL实战_用户在线分析_需求分析

156.SPARKSQL实战_用户在线分析_错位关联

157.SPARKSQL实战_用户在线分析_数据补全和过滤

158.SPARKSQL实战_用户在线分析_总时长_次数_最大时长

159.SPARKSQL实战_用户在线分析_每小时在线人数一

160.SPARKSQL实战_用户在线分析_每小时在线人数二

161.SPARKSQL实战_用户在线分析_每小时在线人数三

162.SPARKSQL实战_用户在线分析_每小时在线人数四