文档章节

sparkSql catalyst优化器

请输入昵称被使用了
 请输入昵称被使用了
发布于 01/14 15:34
字数 1921
阅读 50
收藏 0

相关概念

  • AST树 SQL语法树是编译后被解析的树状结构,树包括很多对象,每个节点都有特定的数据类型,同事有孩子节点(TreeNode对象)。

  • 规则 等价规则转化将规则用于语法树。任何一个SQL优化器中,都会定义大量的Rule,SQL优化器遍历所有节点。匹配所有给定规则,如果匹配成功进行相应转换;失败则继续遍历下一个节点。

Catalyst工作流程

  • Parser,利用ANTLR将sparkSql字符串解析为抽象语法树AST,称为unresolved logical plan/ULP
  • Analyzer,借助于数据元数据catalog将ULP解析为logical plan/LP
  • Optimizer,根据各种RBO,CBO优化策略得到optimized logical plan/OLP,主要是对Logical Plan进行剪枝,合并等操作,进而删除掉一些无用计算,或对一些计算的多个步骤进行合并

优化分类

基于规则优化/Rule Based Optimizer/RBO

  • 一种经验式、启发式优化思路,基于已知的优化方法定义具体的规则进行优化。
  • 对于核心优化算子join有点力不从心,如两张表执行join,到底使用broadcaseHashJoin还是sortMergeJoin,目前sparkSql是通过手工设定参数来确定的,如果一个表的数据量小于某个阈值(默认10M)就使用broadcastHashJoin

常见的三种规则优化:谓词下推、常量累加、列剪枝

  • 谓词下推:扫描数据量过滤,比如合并两个过滤条件为一个减少一次过滤扫描
  • 常量累加:减少常量操作,eg:从100+80优化为180避免每一条record都需要执行一次100+80的操作
  • 列剪枝(去掉不需要的列):对列式数据库提高扫描效率,减少网络、内存数据量消耗

基于代价优化/Cost Based Optimizer/CBO

许多基于规则的优化技术已经实现,但优化器本身仍然有很大的改进空间。例如,没有关于数据分布的详细列统计信息,因此难以精确地估计过滤(filter)、连接(join)等数据库操作符的输出大小和基数 (cardinality)。由于不准确的估计,它经常导致优化器产生次优的查询执行计划,此时就需要基于代价(io和cpu的消耗)的优化器,通过对表数据量等进行优化。

Spark SQL会依据逻辑执行计划生成至少一个物理执行计划,随后通过Cost Model对每个物理执行计划进行开销评估,并选择预估开销最小的一个作为最终的物理执行计划送去做代码生成。

具体步骤如下。

  1. 首先采集原始表基本信息,例如:表数据量大小,行数,列信息(Max,min,非空,最大列长等),列直方图

  2. 再定义每种算子的基数评估规则,即一个数据集经过此算子执行之后基本信息变化规则。这两步完成之后就可以推导出整个执行计划树上所有中间结果集的数据基本信息

  3. 定义每种算子的执行代价,结合中间结果集的基本信息,此时可以得出任意节点的执行代价

  4. 将给定执行路径上所有算子的代价累加得到整棵语法树的代价

  5. 计算出所有可能语法树代价,并选出一条代价最小的

jion操作优化

  1. 配置自动广播的阈值。spark.sql.autoBroadcastJoinThreshold默认值10M,-1代表不进行广播

  2. 使用Executor广播减少Driver内存压力。默认的BroadCastJoin会将小表的内容,全部收集到Driver中,因此需要适当的调大Driver的内存,当广播任务比较频繁的时候,Driver有可能因为OOM而异常退出。

    此时,可以开启Executor广播,配置Executor广播参数spark.sql.bigdata.useExecutorBroadcasttrue,减少Driver内存压力。

  3. 小表执行超时,会导致任务结束。默认情况下,BroadCastJoin只允许小表计算5分钟,超过5分钟该任务会出现超时异常,而这个时候小表的broadcast任务依然在执行,造成资源浪费。

    这种情况下,有两种方式处理:

    • 调整spark.sql.broadcastTimeout的数值,加大超时的时间限制。
    • 降低spark.sql.autoBroadcastJoinThreshold的数值,不使用BroadCastJoin的优化。
  4. spark内部优化。通过谓词下推和布隆过滤器对jion操作进行优化。

    • 针对每个join评估当前两张表使用每种join策略的代价,根据代价估算确定一种代价最小的方案

    • 不同physical plans输入到代价模型(目前是统计),调整join顺序,减少中间shuffle数据集大小,达到最优输出

    • 详见...

扩展数据源优化

扩展数据源是指mySql,Hdfs等数据源,sparkSql针对外部数据源,对查询逻辑进行优化,使得尽可能少的数据被扫描到spark内存中。

例如:SELECT * FROM MYSQL_TABLE WHERE id > 100在没有优化前,会将表里面的所有数据先加载到spark内存中,然后进行筛选。在经过扩展数据源优化后,where 后面的过滤条件会在mysql中执行。

执行计划查看

// 物理逻辑计划,包括parsed logical plan,Analyzed Logical plan,Optimized logical plan
spark.sql("select * from tab0 ").queryExecution
// 查看物理执行计划
spark.sql("select * from tab0 ").explain
// 使用Spark WebUI进行查看

举例:

// 定义DF
scala> val df = park.read.option("header","true").csv("file:///user/iteblog/sales.csv")
df: org.apache.spark.sql.DataFrame = [transactionId: string, customerId: string ... 2 more fields]
// 给 amountPaid 列乘以1
scala> val multipliedDF = df.selectExpr("amountPaid * 1")
multipliedDF: org.apache.spark.sql.DataFrame = [(amountPaid * 1): double]
// 查看优化计划 
scala> println(multipliedDF.queryExecution.optimizedPlan.numberedTreeString)
00 Project [(cast(amountPaid#89 as double) * 1.0) AS (amountPaid * 1)#91]
01 +- Relation[transactionId#86,customerId#87,itemId#88,amountPaid#89] csv

Spark中的所有计划都是使用tree代表的。所以numberedTreeString方法以树形的方式打印出优化计划。正如上面的输出一样。

  所有的计划都是从下往上读的。下面是树中的两个节点:   1、01 Relation:表示从csv文件创建的DataFrame;   2、00 Project:表示投影(也就是需要查询的列)。

从上面的输出可以看到,为了得到正确的结果,Spark通过castamountPaid转换成double类型。

自定义优化器

继承RuleExecutor类,实现其apply方法。不需要像UDF一样注册,默认碰到此场景会执行此优化规则。

Catalyst是一个单独的模块类库,这个模块是基于规则的系统。这个框架中的每个规则都是针对某个特定的情况来优化的。比如:ConstantFolding规则用于移除查询中的常量表达式。

Spark 2.0提供了API,我们可以基于这些API添加自定义的优化规则。实现可插拔方式的Catalyst规则添加。

object MultiplyOptimizationRule extends RuleExecutor[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
       case Multiply(left,right) if right.isInstanceOf[Literal] &&
           right.asInstanceOf[Literal].value.asInstanceOf[Double] == 1.0 =>
           println("optimization of one applied")
           left
       }
}

参考:

CBO详细规则:https://issues.apache.org/jira/secure/attachment/12823839/Spark_CBO_Design_Spec.pdf

华为给spark2.2.0提供了: A cost-based optimizer framework for Spark SQL

在Spark SQL中定义查询优化规则:https://www.iteblog.com/archives/1706.html#Catalyst

SparkSQL Catalyst简介:http://hbasefly.com/2017/03/01/sparksql-catalyst/

Cost Based Optimizer in Apache Spark 2.2:https://databricks.com/blog/2017/08/31/cost-based-optimizer-in-apache-spark-2-2.html

© 著作权归作者所有

请输入昵称被使用了
粉丝 1
博文 67
码字总数 113687
作品 0
朝阳
程序员
私信 提问
sparksql执行流程分析

在前面的文章《spark基础(上篇)》和《spark基础(下篇)》里面已经介绍了spark的一些基础知识,知道了spark sql是spark中一个主要的框架之一。本文我们通过源码,来介绍下spark sql的执行流...

ZPPenny
2017/03/27
0
0
Spark(三) -- Shark与SparkSQL

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/qq1010885678/article/details/45726665 首先介绍一下Shark的概念 Shark简单的说就是Spark上的Hive,其底层依...

jchubby
2015/05/14
0
0
Spark原理 | SparkSQL Catalyst解析

Catalyst Optimizer是SparkSQL的核心组件(查询优化器),它负责将SQL语句转换成物理执行计划,Catalyst的优劣决定了SQL执行的性能。 查询优化器是一个SQL引擎的核心,开源常用的有Apache Calc...

封神
2018/11/27
0
0
大数据(Spark-S3-SparkSQL架构及原理)

Spark SQL的发展 HDFS -> HIVE 由于Hadoop在企业生产中的大量使用,HDFS上积累了大量数据,为了给熟悉RDBMS但又不理解MapReduce的技术人员提供快速上手的工具,Hive应运而生。Hive的原理是将...

这很耳东先生
07/18
45
0
Apache Spark 系列技术直播 - Spark SQL 实践与优化

Apache Spark 系列技术直播 Spark SQL 实践与优化 内容简介: SparkSQL介绍 基本原理 支持的DataSource介绍 Hue/Zepplin/Livy周边跟SparkSQL的集成使用等 SparkSQL优化 SparkSQL Catalyst优化...

开源大数据
2018/11/23
0
0

没有更多内容

加载失败,请刷新页面

加载更多

OSChina 周一乱弹 —— 后来马云就一心想挣钱了

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 小小编辑:《空帆船》- 朴树 《空帆船》- 朴树 手机党少年们想听歌,请使劲儿戳(这里) @webw :第二次被锁在电梯里了 上次你忘带电梯卡, ...

小小编辑
33分钟前
170
7
关于does not give a valid preprocessing token

#define VFUNC(self) ((##self##)->_vptr) 这样在gcc下会编译失败, VC不会 报pasting ) does not give a valid preprocessing token 据说是因为版本问题 解法:去掉## define VFUNC(self) (......

shzwork
34分钟前
3
0
CSS盒子模型

一、什么叫框模型 页面元素皆为框(盒子) 定义了元素框处理元素内容,内边距,外边距以及边框的计算方式 二、外边距 围绕在元素边框外的空白距离(元素与元素之间的距离) 语法:margin,定...

wytao1995
今天
5
0
Replugin借助“UI进程”来快速释放Dex

public static boolean preload(PluginInfo pi) { if (pi == null) { return false; } // 借助“UI进程”来快速释放Dex(见PluginFastInstallProviderProxy的说明) return PluginFastInsta......

Gemini-Lin
今天
4
0
Hibernate 5 的模块/包(modules/artifacts)

Hibernate 的功能被拆分成一系列的模块/包(modules/artifacts),其目的是为了对依赖进行独立(模块化)。 模块名称 说明 hibernate-core 这个是 Hibernate 的主要(main (core))模块。定义...

honeymoose
今天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部