pyspark系列5-Spark SQL介绍

- 日理万妓 2023-01-17 12:52 176阅读 0赞

### 文章目录 ###

*  一.Spark SQL的概述
 *   *  1.1 Spark SQL 来源
     *  1.2 从代码看Spark SQL的特点
     *  1.3 从代码运行速度看来看Spark SQL
 *  二.Spark SQL数据抽象
 *   *  2.1 DataFrame
     *  2.2 Dataset
 *  三.Spark SQL 操作数据库
 *   *  3.1 Spark SQL操作Hive数据库
     *   *  3.1.1 Spark1版本使用方法
         *  3.1.2 Spark2版本使用方法
     *  3.2 Spark SQL操作MySQL数据库
     *   *  3.2.1 Spark1的写法
         *  3.2.2 Spark2的写法
 *  参考:

# 一.Spark SQL的概述 #

## 1.1 Spark SQL 来源 ##

Hive是目前大数据领域，事实上的数据仓库标准。  
![image.png][]

Hive与RDBMS的SQL模型比较类似，容易掌握。 Hive的主要缺陷在于它的底层是基于MapReduce的，执行比较慢。

在Spark 0.x版的时候推出了Shark，Shark与Hive是紧密关联的，Shark底层很多东西还是依赖于Hive，修改了内存管理、物理计划、执行三个模块，底层使用Spark的基于内存的计算模型，性能上比Hive提升了很多倍。

在Spark 1.x的时候Shark被淘汰。在2014 年7月1日的Spark Summit 上， Databricks宣布终止对Shark的开发，将重点放到 Spark SQL 上。

**Shark终止以后，产生了两个分支:**

1). Hive on Spark  
hive社区的，源码在hive中

2). Spark SQL(Spark on Hive)  
Spark社区，源码在Spark中，支持多种数据源，多种优化技术，扩展性好很多;

Spark SQL的源码在Spark中，而且新增了许多的优化代码，如果追求速度，例如数据分析的时候，可以使用Hive on Spark，如果追求性能，例如生产的定时报表的时候，应该使用Spark SQL。

## 1.2 从代码看Spark SQL的特点 ##

我们来对比Spark RDD、Dataframe、SQL代码实现wordcount:  
![image.png][image.png 1]

我们可以看到，Spark SQL代码看起来与关系型数据库是一致的，从上图可以看到Spark SQL的特点:  
**1). 集成**  
通过Spark SQL或DataFrame API运行Spark 程序,操作更加简单、快速.

![image.png][image.png 2]

**从上图可以看到，Spark SQL和DataFrame底层其实就是调用RDD**

**2). 统一的数据访问**  
DataFrame 和SQL提供了访问各种数据源的通用方式，包括Hive、Avro、Parquet、ORC、JSON和JDBC。您甚至可以跨这些数据源连接数据。

![image.png][image.png 3]

**3). Hive集成**  
在现有的数据仓库上运行SQL或HiveQL查询。

**4). 标准的连接**  
服务器模式为业务智能工具提供行业标准的JDBC和ODBC连接。

## 1.3 从代码运行速度看来看Spark SQL ##

![；][07bda9c8fc39079c13f57fda2817424d.png]

从上图我们可以看到:  
1). Python操作RDD比Java/Scala慢一倍以上  
2). 无论是那种语言操作DataFrame，性能几乎一致

**那么为什么Python用RDD这么慢?**  
为什么用Python写的RDD比Scala慢一倍以上，两种不同的语言的执行引擎，上下文切换、数据传输。

Spark SQL其实底层调用的也是RDD执行，其实中间的执行计划进行了优化，而且是在Spark的优化引擎里面,所以无论是那种语言操作DataFrame，性能几乎一致

# 二.Spark SQL数据抽象 #

Spark SQL提供了两个新的抽象，分别是DataFrame 和Dataset；

Dataset是数据的分布式集合。Dataset是Spark 1.6中添加的一个新接口，它提供了RDDs的优点(强类型、使用强大lambda函数的能力)以及Spark SQL优化的执行引擎的优点。可以从JVM对象构造数据集，然后使用函数转换(map、flatMap、filter等)操作数据集。数据集API可以在Scala和Java中使用。Python不支持Dataset API。但是由于Python的动态特性，Dataset API的许多优点已经可以使用了(例如，您可以通过名称natural row. columnname访问行字段)。R的情况也是类似的。

DataFrame 是组织成命名列的Dataset。它在概念上相当于关系数据库中的表或R/Python中的数据框架，但在底层有更丰富的优化。数据框架可以从各种各样的数据源构建，例如:结构化数据文件、Hive中的表、外部数据库或现有的rdd。DataFrame API可以在Scala、Java、Python和r中使用。在Scala和Java中，DataFrame是由行数据集表示的。在Scala API中，DataFrame只是Dataset\[Row\]的类型别名。而在Java API中，用户需要使用Dataset来表示DataFrame。

## 2.1 DataFrame ##

DataFrame的前身是SchemaRDD。Spark1.3更名为DataFrame。不继承RDD，自己实现RDD的大部分功能。

与RDD类似，DataFrame也是一个分布式数据集  
1). DataFrame可以看做分布式Row对象的集合，提供了由列组成的详细模式信息，使其可以得到优化，DataFrame不仅有比RDD更多的算子，还可以进行执行计划的优化

2). DataFrame更像传统数据库的二维表格，除了数据以外，还记录数据的结构信息，即schema

3). DataFrame也支持嵌套数据类型（struct、array和Map）

4). DataFrame API提供的是一套高层的关系操作，比函数式RDD API更加优化，门槛低

5). DataFrame的劣势在于在编译期缺少类型安全检查，导致运行时出错。

![image.png][image.png 4]

## 2.2 Dataset ##

Dataset时在Spark1.6中添加的新接口；与RDD相比，可以保存更多的描述信息，概念上等同于关系型数据库中的二维表。与DataFrame相比，保存了类型信息，是强类型，提供了编译时检查。

调用Dataset的方法会生成逻辑计划，然后Spark的优化器进行优化，最终胜出无力计划，然后提交到集群中运行。

Dataset包含了DataFrame的功能，在Spark2.0中两者得到了统一，DataFrame表示为Dataset\[Row\]，即Dataset的子集.

![image.png][image.png 5]

# 三.Spark SQL 操作数据库 #

## 3.1 Spark SQL操作Hive数据库 ##

### 3.1.1 Spark1版本使用方法 ###

列举一个pyspark运行Spark SQL的案例。

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import HiveContext,SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    hive_context= HiveContext(spark )
    
    # 拼接sql语句
    hive_read = "select * from test.emp"
    
    # 运行sql语句
    read_df = hive_context.sql(hive_read)
    
    # 打印列头及字段
    print("\n\n\n")
    print(read_df)
    
    # 打印查询得到的数据
    print("\n\n\n\n\n")
    read_df.show()
    
    # 关闭spark回话
    spark.stop()

**测试记录:**  
print(read\_df) 输出:  
![image.png][image.png 6]

read\_df.show() 输出:  
![image.png][image.png 7]

### 3.1.2 Spark2版本使用方法 ###

列举一个pyspark运行Spark SQL的案例。  
**代码:**

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行的就直接是DataFrame类型了
    df1 = spark.sql("select * from test.emp")
    
    # 打印列头及字段类型
    print("\n\n\n")
    print(df1)
    
    # 打印查询得到的数据
    print("\n\n\n\n\n")
    df1.show()
    
    # 关闭spark回话
    spark.stop()

**测试记录:**  
![image.png][image.png 8]

![image.png][image.png 9]

## 3.2 Spark SQL操作MySQL数据库 ##

Spark SQL不仅可以操作Hive数据库，也可以操作远程的MySQL数据库

### 3.2.1 Spark1的写法 ###

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import HiveContext,SparkSession
    from pyspark.sql import SQLContext
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    ctx = SQLContext(spark)
    
    jdbcDf=ctx.read.format("jdbc").options(url="jdbc:mysql://10.31.1.123:3306/test",
                                           driver="com.mysql.jdbc.Driver",
                                           dbtable="(SELECT * FROM EMP) tmp",user="root",
                                           password="abc123").load()
    # 打印列头及字段类型
    print(jdbcDf.printSchema())
    
    # 打印查询得到的数据
    print(jdbcDf.show())
    
    # 关闭spark回话
    spark.stop()

**测试记录:**  
print(jdbcDf.printSchema()) 输出:  
![image.png][image.png 10]

print(jdbcDf.show()) 输出:  
![image.png][image.png 11]

### 3.2.2 Spark2的写法 ###

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    jdbcDf=spark.read.format("jdbc").options(url="jdbc:mysql://10.31.1.123:3306/test",
                                           driver="com.mysql.jdbc.Driver",
                                           dbtable="(SELECT * FROM EMP) tmp",
                                           user="root",
                                           password="abc123").load()
    # 打印列头及字段类型
    print(jdbcDf.printSchema())
    
    # 打印查询得到的数据
    print(jdbcDf.show())
    
    # 关闭spark回话
    spark.stop()

**测试记录:**  
![image.png][image.png 12]

![image.png][image.png 13]

# 参考: #

1.http://spark.apache.org/sql/  
2.https://www.jianshu.com/p/ad6dc9467a6b  
3.https://blog.csdn.net/u011412768/article/details/93426353  
4.https://blog.csdn.net/luoganttcc/article/details/88791460

[image.png]: /images/20221021/f479470ebd0540e291434a192bc4446e.png
[image.png 1]: /images/20221021/f7d3b20e817e423bbe2b6ec363a75086.png
[image.png 2]: /images/20221021/19155a917066477cbdcaa04c6a8cdb65.png
[image.png 3]: /images/20221021/d64cfd3847b04a9394d7dabb256b1bc7.png
[07bda9c8fc39079c13f57fda2817424d.png]: /images/20221021/950d5d6a88e04572b0ee01d823debbf2.png
[image.png 4]: /images/20221021/e2c05975379d4c2e8037a21a4789b1b6.png
[image.png 5]: /images/20221021/de9d3e29e0574dd88b7844483311933b.png
[image.png 6]: /images/20221021/b81bff7ecbeb43289a88a6c313369e14.png
[image.png 7]: /images/20221021/b3209cd969ee49e5a57b9ef159272a62.png
[image.png 8]: /images/20221021/07e7171809f74fc7877526c5d6884b14.png
[image.png 9]: /images/20221021/74bd8da85dd04d65871e8c2ccb229ad8.png
[image.png 10]: /images/20221021/dfdb73f44e95463aabcc55d014ede8f7.png
[image.png 11]: /images/20221021/86fe37ae5c0547b3939f358eb94e4b63.png
[image.png 12]: /images/20221021/a10f21760f074295a54d59041a58a3d2.png
[image.png 13]: /images/20221021/d74e010d915e417bb7836d2eed83ca9b.png