马上注册,结交更多数据大咖,获取更多知识干货,轻松玩转大数据
您需要 登录 才可以下载或查看,没有帐号?立即注册
x
1 Spark SQL执行计划概述
1.1 4个计划
Parsed Logical Plan Spark使用 ANTLR4来将SQL字符串解析为最初的LogicalPlan。 Analyzed Logical Plan
调用Spark的Analyzer将最初的Parsed Plan转化成分析后的LogicalPlan。 Optimized Logical Plan
将Analyzed Logical Plan调用catalyst内置的优化策略,生成优化后的LogicalPlan。 Physical Plan
使用Spark的物理策略处理优化后的LogicalPlan的每个节点。
1.2 一个例子1.3 Spark SQL目前所做的优化Databricks以及众多Spark代码贡献者为Spark做了许多优化,最著名的,有SQL优化引擎Catalyst,有针对内存、CPU和I/O的Tungsten计划,还有正在逐步完善的CBO。在其中产生了Code Generation技术和Vectorization技术。参考[SPARK-12795]和[SPARK-12992]。 CodeGen 全称是Whole-Stage Code Generation技术,会在运行时动态生成Java代码,避免虚函数调用。 Vectorization
每次调用next()返回一个batch的数据,减少虚函数调用次数。 CBO
基于代价的优化,主要用来减少join的shuffle。
首先还是说一下基本的火山模型。 1.3.1 Volcano火山模型是一般SQL引擎经常使用的一种模型,还是用上文的例子: 其语法树的火山模型如上图,project节点传next()方法给filter节点,filter节点传next()给全表扫描节点,然后全表扫描节点返回元组给tuple,以此类推。可以发现调用了非常多的next()方法。而next()函数又是虚函数,那么什么是虚函数?给出以下几个定义: 可以看到,如果是这种传统的火山模型,效率会非常低。 1.3.2 Whole-Stage Code Generation那么我们再看一个例子: 这个例子当然可以用火山模型来画出来,但是如果换种Java手写代码的方式呢? 可以看到完全不需要next()方法。网上有一张很著名的图,是一张关于火山模型和大学新生手写代码的性能对比: 非常可怕,对不对?Spark为此实现了Whole-Stage Code Generation优化技术。其目的如下: 避免了虚函数调用:生成的Java代码没有任何虚函数调用逻辑。 使用CPU寄存器存取中间数据:在生成的Java代码中,原本返回的数据会作为变量存放在寄存器中,而不是作为返回值放在内存中。存取更快。 编译器Loop Unrolling:手写代码针对某特定功能使用简单循环,而现代的编译器可以自动的对简单循环进行Unrolling,生成单指令多数据流(SIMD),在每次CPU指令执行时处理多条数据。
1.3.3 Vectorization向量化的作用与CodeGen类似: 2 Spark 查询优化之数据源优化一个IoT下的场景:数据原本存于传统RDBMS,有唯一标识id,为了迎接工业4.0,迁移到大数据平台,要求查询性能跟查询RDBMS性能持平,并且还需要一些固定查询做ETL和报表。 那么IoT数据有什么特征? 我们没有用HBase和Impala,为什么呢?因为我们的数据 没有Update、Delete操作和多版本需求,这是HBase的一大特性; 我们需要建立跟RDBMS看起来一样的Table使用Spark SQL查询。 有很多基于标识字段和时间字段的混合查询。 Impala是MPP架构,适合即时查询,不适合批量查询,而我们有ETL job。
所以最终我们选择的方案是直接存到HDFS上: 使用Parquet Spark推荐的、其支持非常好的高性能列式存储格式,Spark SQL也为其实现了向量化。并且Parquet支持min/max,同时也会将count等信息存入其meta中。 使用Spark分区表
将数据最常见使用场景中的关键字段作为分区字段,使用一个分区UDF或者直接使用该分区字段作为分区值,将数据根据分区按照时间顺序存放在HDFS。 实现谓词下推
Parquet结合Spark可以实现分区过滤。将查询条件里的分区字段下推求UDF后的分区,实现分区过滤,减少I/O。此外,通过Parquet的min/max过滤,进一步减少开销。
假设有一个titem表有三个字段,id,name,type,其中type是ttype表的主键,那么我们可以用type外键来当作分区字段,分区前的schema如下: { "type" : "struct", "fields" : [{ "name" : "id", "type" : "string", "nullable" : false, "metadata" : {} },{ "name" : "name", "type" : "string", "nullable" : true, "metadata" : {} },{ "name" : "type", "type" : "int", "nullable" : false, "metadata" : {} }
分区后的schema如下: partitionColumn: { "name" : "type", "type" : "int", "nullable" : false, "metadata" : {} }
dataColumns: { "type" : "struct", "fields" : [{ "name" : "id", "type" : "string", "nullable" : false, "metadata" : {} },{ "name" : "name", "type" : "string", "nullable" : true, "metadata" : {} }
存到HDFS上的路径就会是如下图的格式: 在这种情况下,用Spark来读写分别是:
当然,实际情况下我们并不会用这种方式来写,因为这种方式会导致全部shuffle,数据还是通过Kafka或者Flink来导入单独append在每个分区目录的里面的。 这种情况下性能对比如图: 2.1 Vectorization的使用限制Spark源码中,FileFormat接口有一个方法叫supportBatch(),顾名思义,就是能不能使用nextBatch(),也就是向量化的批量返回。那么在ParquetFileFormat实现类里,是这么继承这个方法的: 可以看到,当中有一个参数,叫做 wholeStageMaxNumFields,也就是支持CodeGen的最大字段数量: 可以看到,它的默认值是100,为什么会有这个设置呢?两个原因: 2.2 谓词下推需要继承接口: 3 Spark 查询优化之执行计划优化3.1 Cost-Based Optimization(CBO)基于代价的优化,就是根据实际数据量,评估每个计划的执行代价,选择代价最小的执行计划,主要用于Join优化,减少Shuffle。举一个tpc-ds的例子: 它的语法树是: 也就是说,在CBO开启的情况下,它知道storesales和datedim优先join是最好的,因为date_dim是维度表,数据量非常小,而item的数据量是大的。 使用方法:
意义: 3.2 Aggregate Pushdown聚合下推,曾经号称在Spark2.3就要实现,但是迟迟未上,可能由于其真的难点太多,主要难点有二: 但是在某些特定case下聚合下推还是比较好实现的,比如一个大学Oracle课程中的小例子: 在聚合下推前,执行计划是这样的: 在聚合下推后,执行计划变成了这样: 可以看到,在Join的其中一个分支,加入了一个聚合节点。 在tpc-ds某句SQL中拉下来一个单独的子SQL的性能对比如下: SELECT AVG(cs_ext_discount_amt) FROM catalog_sales, date_dim WHERE d_date BETWEEN '1999-02-22' AND CAST('1999-05-22' AS DATE) AND d_date_sk = cs_sold_date_sk GROUP BY cs_sold_date_sk;
由于下推到的数据源本身并没有实现聚合功能,所以相当于推到Spark的driver端后我们自己写代码计算了一遍聚合,如果在数据源本身就有聚合结果的话,查询时间是完全可以在10秒以内的。
|