PySpark（四）PySpark SQL、Catalyst优化器、Spark SQL的执行流程

PySpark SQL

基础

SparkSession对象

DataFrame入门

DataFrame构建

DataFrame代码风格

DSL

SQL

SparkSQL Shuffle 分区数目

DataFrame数据写出

Spark UDF

Catalyst优化器

Spark SQL的执行流程

PySpark SQL

基础

PySpark SQL与Hive的异同

Hive和Spark 均是:“分布式SQL计算引擎”
均是构建大规模结构化数据计算的绝佳利器，同时SparkSQL拥有更好的性能。
目前，企业中使用Hive仍旧居多，但SparkSQL将会在很近的未来替代Hive成为分布式SQL计算市场的顶级

这里的重点是：Spark SQL能支持SQL和其他代码混合执行，自由度更高，且其是内存计算，更快。但是其没有元数据管理，然而它最终还是会作用到Hive层面，可以调用Hive的Metasotre

SparkSQL的基本对象是DataFrame，其特点及与其他对象的区别为：

SparkSQL 其实有3类数据抽象对象

SchemaRDD对象 (已废弃)
DataSet对象: 可用于Java、Scala语言
DataFrame对象:可用于Java、Scala、Python、R

SparkSession对象

在RDD阶段，程序的执行入口对象是: SparkContext
在Spark 2.0后，推出了SparkSession对象，作为Spark编码的统一入口对象
SparkSession对象可以:
-用于SparkSQL编程作为入口对象
- 用于SparkCore编程，可以通过SparkSession对象中获取到SparkContext

from pyspark.sql import SparkSession
if __name__ == '__main__':
    spark =  SparkSession.builder.appName('lmx').master('local[*]').getOrCreate()
    sc = spark.sparkContext

DataFrame入门

DataFrame的组成如下
在结构层面
StructType对象描述整个DataFrame的表结构

StructField对象描述一个列的信息
在数据层面
Row对象记录一行数据
Column对象记录一列数据并包含列的信息

DataFrame构建

1、用RDD进行构建

rdd的结构要求为：[[xx,xx],[xx,xx]]

spark.createDataFrame(rdd,schema=[])

    spark =  SparkSession.builder.appName('lmx').master('local[*]').getOrCreate()
    sc = spark.sparkContext
    rdd = sc.textFile('data/input/sql/people.txt').map(lambda x:x.split(',')).map(lambda x:[x[0],int(x[1])])
    print(rdd.collect())
    # [['Michael', 29], ['Andy', 30], ['Justin', 19]]
    df = spark.createDataFrame(rdd,schema=['name','age'])
    df.printSchema()#打印表结构
    df.show()#打印表
#     root
#     | -- name: string(nullable=true)
#     | -- age: long(nullable=true)
# 
# +-------+---+
# | name | age |
# +-------+---+
# | Michael | 29 |
# | Andy | 30 |
# | Justin | 19 |
# +-------+---+

2、利用StructType进行创建

需要先引入StructType,StringType,IntegerType等构建schema

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType,StringType,IntegerType
if __name__ == '__main__':
    spark =  SparkSession.builder.appName('lmx').master('local[*]').getOrCreate()
    sc = spark.sparkContext
    rdd = sc.textFile('data/input/sql/people.txt').map(lambda x:x.split(',')).map(lambda x:[x[0],int(x[1])])
#构建schema    
schema =StructType().add("name",StringType(),nullable=False).\
        add('age',IntegerType(),nullable=True)
    df = spark.createDataFrame(rdd,schema=schema)
    df.printSchema()
    df.show()

3、toDF将rdd转换为df

下面展示了两种方式

    # 只设定列名，列的数据结构则是内部自己判断
    df = rdd.toDF(['name','age'])
    df.printSchema()
    # root
    # | -- name: string(nullable=true)
    # | -- age: long(nullable=true)
    # 设定列名和数据类型
    schema =StructType().add("name",StringType(),nullable=False).\
        add('age',IntegerType(),nullable=True)
    df = rdd.toDF(schema=schema)
    df.printSchema()
    # root
    # | -- name: string(nullable=false)
    # | -- age: integer(nullable=true)

4、基于pandas构建

    dfp = pd.DataFrame({
        "id":[1,2,3],
        'score':[99,98,100]
    })
    df = spark.createDataFrame(dfp)
    df.printSchema()
    df.show()
    # root
    # | -- id: long(nullable=true)
    # | -- score: long(nullable=true)
    # 
    # +---+-----+
    # | id | score |
    # +---+-----+
    # | 1 | 99 |
    # | 2 | 98 |
    # | 3 | 100 |
    # +---+-----+

5、通过文件读取创造

在读取json和parquet文件时不需要设定schema，因为文件已经自带

而读取csv时，还需要使用.option设定 header等参数

这里说一下parquet文件

parquet:是Spark中常用的一种列式存储文件格式
和Hive中的ORC差不多，他俩都是列存储格式
parquet对比普通的文本文件的区别：

parquet 内置schema(列名列类型是否为空)
存储是以列作为存储格式
存储是序列化存储在文件中的(有压缩属性体积小)

DataFrame代码风格

DataFrame支持两种风格进行编程，分别是DSL风格和SQL风格
DSL语法风格
DSL称之为:领域特定语言
其实就是指DataFrame的特有API
DSL风格意思就是以调用API的方式来处理Data比如: df.where0.limit0
SQL语法风格
SQL风格就是使用SQL语句处理DataFrame的数据比如: spark.sql(“SELECT*FROM xxx)

DSL

其实就是用其内置的API处理数据，举例：

    df.select('id','subject').show()
    df.where('subject="语文"').show()
    df.select('id','subject').where('subject="语文"').show()
    df.groupBy('subject').count().show()

API其实跟SQL类似，这里不详细说明了，个人感觉不如直接写SQL语句

SQL

DataFrame的一个强大之处就是我们可以将它看作是一个关系型数据表，然后可以通过在程序中使用spark.sgl0来执行SQL语句查询，结果返回一个DataFrame。如果想使用SQL风格的语法，需要将DataFrame注册成表采用如下的方式:

    df.createTempView('tmp') #创建临时视图
    df.createGlobalTempView('global_tmp')#创建全局试图
    # 全局表: 跨SparkSession对象使用在一个程序内的多个SparkSession中均可调用查询前带上前缀：global_tmp
    df.createOrReplaceTempView('repalce_tmp')#创建临时表，如果存在则替换

然后使用spark.sql的形式书写sql代码

    spark.sql('select * from tmp where subject = "语文"').show()
    spark.sql('select id,score from repalce_tmp where score>90').show()
    spark.sql('select subject,max(score) from global_temp.global_tmp group by subject').show()

SparkSQL Shuffle 分区数目

原因: 在SparkSQL中当Job中产生Shufle时，默认的分区数 spark.sql.shufle,partitions 为200，在实际项目中要合理的设置。
在代码中可以设置：

spark =  SparkSession.builder.appName('lmx').\
master('local[*]').config('spark.sql.shufle,partitions',2).\
getOrCreate()

spark.sqL.shuffle.partitions 参数指的是，在sql计算中，shuffle算子阶段默认的分区数是200

对于集群模式来说，200个默认也算比较合适

如在Local下运行，200个很多，在调度上会带宋限外的损耗，所以在Local下建议修改比较低，比如2\4\10均可，这个参数和Spark RDD中设置并行度的参数是相互独立的

DataFrame数据写出

统一API：

下面提供两种方法，分别写出为json和csv

    spark.sql(
        'select user_id,avg(score) avg_score from tmp group by user_id order by avg_score desc'
    ).write.mode('overwrite').format('json').save('data/output/1t')

    spark.sql(
        'select user_id,avg(score) avg_score from tmp group by user_id order by avg_score desc'
    ).write.mode('overwrite').format('csv')\
        .option('header',True)\
        .option('sep',';')\
        .save('data/output/csv')

其他的一些方法：

SparkSQL中读取数据和写出数据 - 知乎

不过这里似乎不能自己命名导出的数据文件

Spark UDF

无论Hive还是SparKSQL分析处理数据时，往往需要使用函数，SparkSQL模块本身自带很多实现公共功能的函数，在pyspark.sql.functions中SparkSQL与Hive一样支持定义函数:UDF和UDAF，尤其是UDF函数在实际项目中使用最为广泛。回顾Hive中自定义函数有三种类型:
第一种:UDF(User-Defined-Function)函数.
一对一的关系，输入一个值经过函数以后输出一个值;
在Hive中继承UDF类，方法名称为evaluate，返回值不能为void，其实就是实现一个方法;

第二种:UDAF(User-Defined Aggregation Function)聚合函数

多对一的关系，输入多个值输出一个值，通常与groupBy联合使用;

第三种:UDTF(User-DefinedTable-Generating Functions)函数

一对多的关系，输入一个值输出多个值(一行变为多行)，用户自定义生成函数，有点像flatMap;

在SparkSQL中，目前仅仅支持UDF函数和UDAF函数，目前Python仅支持UDF

UDF有两种定义方式

方式1语法
udf对象=sparksession.udfregister(参数1，参数2，参数3)

参数1:UDF名称，可用于SQL风格

参数2:被注册成UDF的方法名
参数3:声明UDF的返回值类型

udf对象:返回值对象，是一个UDF对象，可用于DSL风格
方式2语法

from pyspark.sql import functions as F

udf对象 = F.udf(参数1，参数2)

参数1:被注册成UDF的方法名

参数2:声明UDF的返回值类型

udf对象:返回值对象，是一个UDF对象，可用于DSL风格

举例：

    def double_score(num):
        return 2*num

    udf1 = spark.udf.register('udf_1',double_score,IntegerType())
    # dsl风格
    df.select(udf1(df['score'])).show()
    # sql风格
    df.selectExpr('udf_1(score)').show()
    # sql风格2
    df.createTempView('tmp')
    spark.sql("select udf_1(score) from tmp").show()

    udf2 = F.udf(double_score,IntegerType())
    df.select(udf2(df['score'])).show()

当返回值是数组时，需要定义数组内部数据的数据类型：ArrayType(StringType())

    spark =  SparkSession.builder.appName('lmx').master('local[*]').config('spark.sql.shufle,partitions',2).getOrCreate()
    sc = spark.sparkContext

    rdd=sc.parallelize([['i love you'],['i like you']])
    df = rdd.toDF(['ifo'])
    def func(num):
        return num.split(' ')
    udf = spark.udf.register('udf_sql',func,ArrayType(StringType()))

    # dsl风格
    df.select(udf(df['ifo'])).show()

当返回值是字典时，需要使用StructType()，且定义每个列的名字（需要跟函数返回值的列名一样）和数据类型

    rdd=sc.parallelize([[1],[2],[3],[4],[5]])
    df = rdd.toDF(['ifo'])
    df.show()
    def func(num):
        return {'num':num,'num1':num+10}
    udf = spark.udf.register('udf_sql',func,StructType().\
                             add('num',IntegerType(),nullable=False).\
                             add('num1',IntegerType(),nullable=False))
    df.select(udf(df['ifo'])).show()

Catalyst优化器

RDD的执行流程为：

代码 ->DAG调度器逻辑任务 ->Task调度器任务分配和管理监控 ->Worker干活

SparkSQL会对写完的代码，执行“自动优化”，既Catalyst优化器，以提升代码运行效率，避免开发者水平影响到代码执行效率。（RDD代码不会，是因为RDD的数据对象太过复杂，无法被针对性的优化）

加入优化的SparkSQL大致架构为：

1.API 层简单的说就是 Spark 会通过一些 API 接受 SQL 语句

2.收到 SQL 语句以后,将其交给 Catalyst,Catalyst 负责解析 SQL,生成执行计划等

3.Catalyst 的输出应该是 RDD 的执行计划

4.最终交由集群运行

Catalyst优化器主要分为四个步骤

1、解析sql，生成AST(抽象语法树)

2、在 AST 中加入元数据信息,做这一步主要是为了一些优化,例如 col=col 这样的条件

以上面的图为例：

score.id → id#1#L 为 score.id 生成 id 为1,类型是 Long
score.math_score→math_score#2#L为 score.math_score 生成 id 为 2,类型为 Long
people.id→id#3#L为 people.id 生成 id 为3,类型为 Long
people.age→age#4#L为 people.age 生成 id 为 4,类型为 Long

3、对已经加入元数据的 AST,输入优化器,进行优化,主要包含两种常见的优化：

谓词下推(Predicate Pushdown)\ 断言下推:将逻辑判断提前到前面,以减少shuffle阶段的数据量。

以上面的demo举例，可以先进行people.age>10的判断再进行Join等操作。

列值裁剪(Column Pruning):将加载的列进行裁剪,尽量减少被处理数据的宽度

以上面的demo举例，由于只select了score和id，所以开始的时候，可以只保留这两个列，由于parquet是按列存储的，所以很适合这个操作

4、上面的过程生成的 AST 其实最终还没办法直接运行,这个 AST 叫做逻辑计划,结束后,需要生成物理计划,从而生成 RDD 来运行

Spark SQL的执行流程

如此，Spark SQL的执行流程为：

1.提交SparkSQL代码
2.catalyst优化
        a.生成原始AST语法数
        b.标记AST元数据
        c.进行断言下推和列值裁剪以及其它方面的优化作用在AST上
        d.将最终AST得到,生成执行计划
        e.将执行计划翻译为RDD代码
3.Driver执行环境入口构建(SparkSession)
4.DAG 调度器规划逻辑任务
5.TASK 调度区分配逻辑任务到具体Executor上工作并监控管理任务
6.Worker干活