Spark Sql Execution【黑马大数据培训】

更新时间:2019年07月29日 15时43分29秒来源:黑马程序员论坛

关于spark-sql模块

spark-sql模块下的代码作用有:

sparkSession，SQLContext，DataSet，DataFrameWriter和reader等类
api 包: python，r的api
catalog包: catalog以及column，table，function，database的接口类
expression包: 包括aggregator，UDF， UDAF以及窗口函数
internal包：包括catalogImpl， HiveSerde，sessionState，sharedState等internal类
jdbc包：里面都是方言类 dialect
sources包: 用于下推至DataSource的filter以及一些DataSource和sql relation相关接口
streaming包: DataStreamReader， writer，StreamingQueryException等于streaming有关的类。
util包: QueryExecutionListener类
execution包：本文的重点

execution包

该包下源码分为:

直接在根目录下
- 基本物理操作: coalesceExec, filterExec, projectExec, unionExec, RangeExec, and etc.
- cacheManager：用于cache table.
- 一些exec: sort, DataSourceScan
- wholeStageCodegenExec
- SparkPlanner用于生产物理计划的
- limit以及其他操作
aggregate包: 当然是与聚合相关的exec 以及UDAF
arrow包: arrow也是一种列式存储格式，这个包有他的 writer以及工具类.
columnar包: 与列状态，访问，类型相关的类
command包: 一些命令,ddl, analyzeTableCommand, functions, create等命令
DataSources包: 与DataSource有关，parquet, jdbc, orc等等
- 关于DataSource options， writer，工具类等等
exchange：里面有broadcastExchangeExec， exchangeCoordinator，shuffleExchangeExec等相关类，后面会重点分析这个,其实在物理计划中转换的重点就是这部分，所以ensureRequirements就在这个包中。
joins: join相关的， hashJoin, broadcastHashJoin, broadcastNestedLoopJoin, sortMergeJoin,shuffleHashJoinExec.
streaming包: 应该是continuous streaming相关的底层实现，应该不是走rdd，是一个真正流式的实现.
window包: 窗口函数相关exec。
python包，r包，metric包。

下文重点分析 aggregate, exchange, joins包。

exchange包Exchange & ExchangeCoordinator

首先讲一下Exchange，顾名思义就是交换，是为了在多个线程之间进行数据交换，完成并行。

Exchange分为两种，一种是BroadcastExchange另外一种是ShuffleExchange。Broadcast就是将数据发送至driver，然后由driver广播，这适合于数据量较小时候的shuffle。另一种ShuffleExchange就比较常见了，就是多对多的分发。

如果开启了spark.sql.adaptive.enabled，也就是自适应执行，

那么在使用ShuffleExchange的时候有对应的ExchangeCoordinator；如果没开启ae，那就不需要协调器。

ExchangeCoordinator顾名思义，是Exchange协调器，是一个用于决定怎么在stage之间进行shuffle数据的coordinator。这个协调器用于决定之后的shuffle有多少个partition用来需要fetch shuffle 数据。

一个coordinator有三个参数.

numExchanges
targetPostShuffleInputSize
minNumPostShufflePartitions

第一个参数是用于表示有多少个ShuffleExchangeExec需要注册到这个coordinator里面。因此，当我们要开始真正执行时，我们需要知道到底有多少个ShuffleExchangeExec。

第二个参数是表示后面shuffle阶段每个partition的输入数据大小，这是用于adaptive-execution的，用于推测后面的shuffle阶段需要多少个partition。可以通过spark.sql.adaptive.shuffle.targetPostShuffleInputSize来配置。

第三个参数是一个可选参数，表示之后shuffle阶段最小的partition数量。如果这个参数被定义，那么之后的shuffle阶段的partition数量不能小于这个值。

Coordinator的流程如下：

在一个SparkPlan执行之前，对于一个ShuffleExchangeExec操作，如果它被指定了一个coorinator，那么它将会注册到这个协调器，这发生在doPrepare方法里.
当开始执行SparkPlan，注册到这个协调器里面的ShuffleExchangeExec将会调用postShuffleRDD方法来相应的 post-shuffle ShuffledRowRDD.如果这个协调器已经决定了如何去shuffle data，那么这个Exec会马上获得它对应的ShuffledRowRDD.
如果这个coordinator已经决定了如何shuffle data，它会让注册到自己的ShuffleExchangeExecs来提交pos-shuffle stage。然后基于pre-shuffle阶段partition的统计信息，来决定post-shuffle的partition数量，如果post-shuffle需要，它也会将一些需要的连续的partitions放在一起发送给post-shuffle.
最后，这个coordinator会为所有注册的ShuffleExchangeExec创建post-shuffle ShuffledRowRDD。

目前的ae是比较老版本的ae，intel有一个ae项目，相信会在spark-3.0之后会合入，可以了解一下新的ae。

EnsureRequirements

这就是前面说的在SparkPlan 之前的do-prepare，需要为sparkplan的可执行做一些准备工作。

主要分为以下几部分:

确定distribution和Ordering
确定join 的条件和join keys出现顺序匹配
创建coordinator
- ae开启
- 支持Coordinator
  - 有ShuffleExchangeExec且是HashPartitionings
  - 支持Distribution并且child个数大于1
- 针对ShuffleExchangeExec创建coordinator
- 针对一个post-shuffle 对应几个pre-shuffle的创建coordinator，例如join 一个对应多个pre-shuffle，而几个pre-shuffle有不同的分区划分方式

Join

这里的join是执行阶段的join，不是解析阶段的join。

join包括BroadcastHashJoinExec, ShuffledHashJoinExec以及SortMergeSortMergeJoinExec。

join策略的选择SparkStrategies类里面，面临join，首先会尝试BroadcastJoin，然后是ShuffledHashJoin最后才是SortMergeHashJoin,只要能满足前面的条件就会优先使用前面的join。

而这些JoinExec是在前面构建物理之前进行构建，在之后真正执行物理计划的时候执行。

Aggregate

Agg和join有些类似，都是需要进行shuffle操作，但不同的是Aggregate可以是一个一元操作，而join是多元操作。

Aggregate操作例如 max, count, min, sum，groupBy， reduce, reduceBy 以及一些UDAF(User Defined Aggregate Function)。而groupBy这些操作可能面临order by.

Conclusion

本文大概讲解了下execution包中各个部分的用途，重点是如何进行Exchange，以及什么是ExchangeCoordinator。对于join和Aggregate未涉及太多。

本文转载来自：https://www.turbofei.wang/spark/2019/05/20/spark-sql-execution

推荐了解热门学科

java培训	Python人工智能	Web前端培训	PHP培训
区块链培训	影视制作培训	C++培训	产品经理培训
UI设计培训	新媒体培训	软件测试培训	Linux运维
大数据培训	智能机器人软件开发

传智播客是一家致力于培养高素质软件开发人才的科技公司，“黑马程序员”是传智播客旗下高端IT教育品牌。自“黑马程序员”成立以来，教学研发团队一直致力于打造精品课程资源，不断在产、学、研3个层面创新自己的执教理念与教学方针，并集中“黑马程序员”的优势力量，针对性地出版了计算机系列教材50多册，制作教学视频数+套，发表各类技术文章数百篇。

传智播客从未停止思考

传智播客副总裁毕向东在2019IT培训行业变革大会提到，“传智播客意识到企业的用人需求已经从初级程序员升级到中高级程序员，具备多领域、多行业项目经验的人才成为企业用人的首选。”

中级程序员和初级程序员的差别在哪里?
项目经验。毕向东表示，“中级程序员和初级程序员最大的差别在于中级程序员比初级程序员多了三四年的工作经验，从而多出了更多的项目经验。“为此，传智播客研究院引进曾在知名IT企业如阿里、IBM就职的高级技术专家，集中研发面向中高级程序员的课程，用以满足企业用人需求，尽快补全IT行业所需的人才缺口。

何为中高级程序员课程?

传智播客进行了定义。中高级程序员课程，是在当前主流的初级程序员课程的基础上，增加多领域多行业的含金量项目，从技术的广度和深度上进行拓展。“我们希望用5年的时间，打造上百个高含金量的项目，覆盖主流的32个行业。”传智播客课程研发总监于洋表示。

黑马程序员热门视频教程

Python入门教程完整版（懂中文就能学会）	零起点打开Java世界的大门
C++\| 匠心之作从0到1入门学编程	PHP\|零基础入门开发者编程核心技术
Web前端入门教程_Web前端html+css+JavaScript	软件测试入门到精通

全国中心

Spark Sql Execution【黑马大数据培训】

相关阅读