手机如何创建网站,wordpress设置手机浏览器,网站开发专员岗位职责,装修网站怎么建设Spark Catalyst逻辑计划逻辑计划解析逻辑计划优化Catalyst 规则优化过程物理计划Spark PlanJoinSelection生成 Physical PlanEnsureRequirementsSpark SQL 端到端的优化流程#xff1a;
Catalyst 优化器 : 包含逻辑优化/物理优化Tungsten :
Spark SQL的优化过程 : 逻辑计划 …
Spark Catalyst逻辑计划逻辑计划解析逻辑计划优化Catalyst 规则优化过程物理计划Spark PlanJoinSelection生成 Physical PlanEnsureRequirementsSpark SQL 端到端的优化流程
Catalyst 优化器 : 包含逻辑优化/物理优化Tungsten :
Spark SQL的优化过程 : 逻辑计划
val userFile: String _
val usersDf spark.read.parquet(userFile)val txFile: String _
val txDf spark.read.parquet(txFile)val users usersDf.select(name, age, userId).filter($age 30).filter($gender.isin(M))val result txDF.select(price, volume, userId).join(users, Seq(userId), inner).groupBy(col(name), col(age)).agg(sum(col(price) * col(volume)).alias(sum)result.write.parquet(_)计算逻辑 : Catalyst 逻辑优化阶段
逻辑计划解析 : 把 Unresolved Logical Plan 换为 Analyzed Logical Plan逻辑计划优化 : 基于启发式规则Heuristics Based Rules) 把 Analyzed Logical Plan 转为 Optimized Logical Plan
Catalyst 逻辑优化阶段 Unresolved Logical Plan : 逻辑计划解析
逻辑计划解析 : 结合 DataFrame 的 Schema 确认计划中的表名、字段名、字段类型和实际数据是否一致。确认后就生成 Analyzed Logical Plan
Analyzed Logical Plan : 逻辑计划优化
同种计算逻辑的多种实现方式
按照不同的顺序对算子做排列组合最好顺序能省则省、能拖则拖的开发原则选择所有实现方式中最优 Catalyst 优化规则范畴 :
谓词下推Predicate Pushdown把谓词 (过滤条件 age 30) 推到离数据源最近列剪裁Column Pruning: 只扫描与查询相关的字段常量替换 Constant Folding: 如 age 12 18 优化成 age 30
Cache Manager 优化
Cache Manager 维护与缓存相关信息。即维护 Mapping 映射字典Key 逻辑计划Value 对应的 Cache 元信息当 Catalyst 进行逻辑计划优化时先在 Cache Manager 查找当该逻辑计划分支在 Cache Manager 时就进行替换该计划
Optimized Logical Plan Catalyst 规则优化过程
逻辑计划Logical Plan物理计划Physical Plan都继承 QueryPlan
QueryPlan 父类 TreeNode
TreeNode 语法树中对节点的抽象TreeNode 有个字段 children 类型是 Seq[TreeNode]利用 TreeNode 类型能构建出树结构
TreeNode 定义了很多高阶函数如transformDown
transformDown 的形参 各种优化规则返回类型是 TreeNodetransformDown 是递归函数先优化当前节点再依次优化 children 中的子节点直到整棵树的叶子节点
transformDown 类似转换过程
//Expression的转换
import org.apache.spark.sql.catalyst.expressions._val myExpr: Expression Multiply(Subtract(Literal(6), Literal(4)), Subtract(Literal(1), Literal(9)))val transformed: Expression myExpr transformDown {// 二元操作符转成加法操作case BinaryOperator(l, r) Add(l, r)// 大于 5 转成 1case IntegerLiteral(i) if i 5 Literal(1)// 小于 5 转成转成 0case IntegerLiteral(i) if i 5 Literal(0)
}转换过程意图 物理计划
物理计划阶段(Physical Planning)
优化 Spark Plan 根据优化策略 (Strategies)把逻辑计划的关系操作符映射成物理操作符生成 Physical Plan 根据 Preparation Rules对 Spark Plan 进行完善 Spark Plan
Spark Plan 优化策略
基于模式匹配的偏函数Partial Functions把逻辑计划中的操作符平行映射为 Spark Plan 中的物理算子
类型优化策略含义作用通用BasicOperators逻辑到物理的基本映射如Project/Filter/SortJoinSelection静态 Joln 策略选择InMemoryScans缓存策略对应逻辑优化阶段的 Cache ManagerAggregation聚合策路Window窗口计算策酪SpecialLimits与 Limit 相关的优化策路PythonEvalsPython UDF 优化策路SparkScriptsTransformation 脚本优化策略StreamingStatefulAggregationStrategy有状态的聚合策略StreamingDeduplicationStrategy流处理中的去重策路StreamingGlobalLimitStrategy流处理中的 Limit 处理策略StreamingJoinStrategy流处理中的 Join 策略StreamingRelationStrategy数据源读取策酪FlatMapGroupsWithStateStrategy流处理中的 FlatMap 优化
JoinSelection
Catalyst 运行时的 Join 策略
Join 策略执行效率排序含义Broadcast Hash Join (BHJ)最优小表构建哈希表把小表广播进行关联Shuffle Sort Merge Join (SMJ)次优先 Shuffle , 再排序进行关联Shuffle Hash Join (SHJ)次优先 Shuffle , 再构建哈希表进行关联Broadcast Nested Loop Join (BNLJ)最差将小表广播进行关联Shuffle Cartesian Product Join (CPJ)最差先 Shuffle 进行关联
数据分发与 Join 实现机制的组合 Join 策略的先决条件
条件型 判决 5 大 Join 策略的先决条件指令型开发者提供的 Join Hints
5 种 Join 策略的先决条件
选择顺序Join 策略Join 类型表大小等值 JoinInner Join不能 Full Outer Join能广播1BHJ√√√2SMJ√3SHJ√4BNLJ√5CPJ√
指令型信息 Join Hints允许个人选择 Join 策略
选择 SHJ
val result txDF.select(price, volume, userId).join(users.hint(shuffle_hash), Seq(userId), inner).groupBy(col(name), col(age)).agg(sum(col(price) * col(volume)).alias(revenue))Spark Plan Join 策略是 SMJ 生成 Physical Plan
从 Spark Plan 到 Physical Plan 的转换需要 Preparation Rules 规则
Preparation Rules
Preparation Rules含义作用EnsureRequirements确保每个操作符的输入要求必要时添加 Shuffle/Sort为 Physical Plan 补充必要的操作保证 Spark Plan 计划的每个步骤能够顺利执行CollapseCodegenStagesTungsten 优化机制全阶段代码生成(Whole Stage Code Generation)在同个 Stage 内部尽可能地把所有操作和计算捏合成一个函数提升计算效率ReuseExchange内存或磁盘中的存储复用同样的执行计划能共享广播变量或 Shuffle 的中间结果避免重复的 Shuffle 操作ReuseSubquery子查询复用复用同样的查询结果避免重复计算PlanSubquery生成子查询对子查询应用 Preparation RulesExtractPythonUDFs提取 Python 的 UDF 函数把 Python UDF 分发到单独的 Python 进程
EnsureRequirements
EnsureRequirements (满足前提条件) : 对执行计划中的每个操作符节点都有 4 个属性用来描述数据输入/ 输出的分布状态
操作符属性含义outputPartitioning输出数据的分区规则outputOrdering输出数据的排序规则requireChildDistribution要求输入数据满足某种分区规则requireChildOrdering要求输入数据满足某种排序规则
Project 不满足 SortMergeJoin 的 Requirements
outputPartitioning 属性 Unknow未 ShuffleoutputOrdering 属性 None 未排序 EnsureRequirements 规则添加 Exchange/Sort
Exchange Shuffle 操作满足 SortMergeJoin 对数据分布的要求Sort 排序满足 SortMergeJoin 对数据有序的要求调用 Physical Plan 的 doExecute 方法把结构化查询的计算结果转换成 RDD[InternalRow]InternalRow Tungsten 设计的定制化二进制数据结构调用 RDD[InternalRow] 上的 Action 算子Spark 就触发 Physical Plan 执行 Physical Plan
EnsureRequirements 在两个分支上添加 Exchange/Sort*(数字) : * : WSCG数字 : Stage 编号数字相同会 WSCG 合成
