Catalyst的工作流程:

• Unresolved Logical Plan:
  • SQL语句首先通过sql parser模块被分词, 形成select, where ,join等语句块, 并将这些语句块行成语法树. 此棵树称为Unresolved Logical Plan
• Logical Plan:
  • 借助表的元数据将Unresolved Logical Plan解析为Logical Plan.
  • 例如, 上一步的逻辑执行框架有了基本骨架后, 系统并不知道tableA包括什么列, 列的数据类型, 表的数据格式(text,orc), 表的物理位置, sql中的函数指的哪个类等等; 这些在信息都保存在标的元数据中, 同时也能判断sql语句的正确性
• Optimized Logical Plan:
  • 通过RBO, 将Logical Plan优化为Optimized Logical Plan
  • 这步优化器是Catalyst最重要的部分(Optimizer类), RBO-指基于规则的优化, 规则有很多种, 常见的有如下3种:
    • 谓词下推:
      比如一个sql:

    select * from A join B on 条件1 where B.age>10;
    

    由于数据量的大小会显著影响join的效率, 因此, 通过谓词下推把join放在where的后面执行. 先用where条件缩小表B的量, 再进行join
    • 常量累加:
      比如一个sql:

    select x+1+2;
    

    查询列会被优化成x+3, 虽然改动很小, 但是优化后每次操作都不会再去执行1+2这个动作
    

    • 列值裁剪
      有的sql语句可能并不会使用到一个表的所有列, 此时只需要用标的部分列参加运算即可, 这尤其在列式存储的表中大大提高扫描性能(减少网络内存的数据量消耗)
• Physical Plan:
  • 得到的Optimized Logical Plan并不能被spark所理解, 此时需要转换为Physical Plan
  • 比如join算子, spark并不知道使用BroadcastJoin, 还是ShuffleHashJoin,还是SortMergeJoin; 物理执行计划就是在这些具体实现中跳出一个耗时最小的算法实现, 这个过程涉及基于代价的优化CBO. join通常有两个选择题要做
    • 一个是上面说的选择哪种join算法
    • 另一个是join的顺序选择
      对于星型模型和雪花模型来讲, 不同的join顺序意味着不同的执行效率. 例如

      A join B join C
      

      A,B表都很大, 但是C表很小, 则AjoinB显然需要大量的系统资源完成; 如果使用'A join C join B'的顺序执行, 因为C很小, 所以A join C会很快得到结果; 而小的结果集再去join B, 性能会显而易见的比前一种方案好. 而这种join顺序的选择, 并没有一个固定的规则来完成, 只有知道表的基础信息(表的大小, 表的记录总条数), 才能从中选择一条代价最小的语法树来执行. 即CBO的核心在于评估处一条给定语法树的实际执行代价

CBO的实现思路

• 采集原始表的基本信息
  这个操作是CBO最基础的一项工作，采集的主要信息包括表级别指标和列级别指标，如下所示，estimatedSize和rowCount为表级别信息，basicStats和Histograms为列级别信息，后者粒度更细，对优化更加重要。
  • estimatedSize: 每个LogicalPlan节点输出数据大小（解压）
  • rowCount: 每个LogicalPlan节点输出数据总条数
  • basicStats: 基本列信息，包括列类型、Max、Min、number of nulls, number of distinct values, max column length, average column length等
  • Histograms: Histograms of columns, i.e., equi-width histogram (for numeric and string types) and equi-height histogram (only for numeric types).

这些指标是一些统计指标, 因此需要单独执行统计, 最好在业务低峰期, 对表数据有较大的变化的表单独统计. hive通过analyze命令对表的数据信息进行统计

• 定义核心算子的基数推导规则
  假如sql中有where条件语句"where cid>N", 如何推导出经过这个条件过滤后的中间表的基本统计信息?
  第一步的原始表基本信息中, 其中有一项是列值分布的Histograms(直方图), 可以从直方图中粗略找到cid大于N的记录有多少条
  <img src="img/cbohistogram,png.png">

• 核心算子实际代价计算
  通常来讲，节点实际执行代价主要从两个维度来定义：CPU Cost以及IO Cost。为后续讲解方便起见，需要先行定义一些基本参数：

  • Hr：从HDFS上读取1byte数据所需代价
  • Hw：往HDFS上写入1byte数据所需代价
  • Tr：数据总条数（the number of tuples in the relation ）
  • Tsz：数据平均大小（Average size of the tuple in the relation ）
  • CPUc：两值比较所需CPU资源代价（CPU cost for a comparison in nano seconds ）
  • NEt：1byte数据通过网络在集群节点间传输花费代价（the average cost of transferring 1 byte over network in the Hadoop cluster from any node to any node ）
  • ……

  上文说过，每种算子的实际执行代价计算方式都不同，挑两个比较简单、容易理解的来分析，第一个是Table Scan算子，第二个是Hash Join算子。

  • Table Scan算子
    直观上来讲这类算子只有IO Cost，CPU Cost为0。Table Scan Cost = IO Cost = Tr * Tsz * Hr，很简单，Tr * Tsz表示需要scan的数据总大小，再乘以Hr就是所需代价。OK，很直观，很简单。
  • Hash Join算子
    以Broadcast Hash Join为例（小表构建hash桶,大表负责探测），假设大表分布在n个节点上，每个节点的数据条数\平均大小分别为Tr(R1)\Tsz(R1)，Tr(R2)\Tsz(R2), … Tr(Rn)\Tsz(Rn)，小表数据条数为Tr(Rsmall)\Tsz(Rsmall)，那么CPU代价和IO代价分别为：
    • CPU Cost = 小表构建Hash Table代价 ＋ 大表探测代价 ＝ Tr(Rsmall) * CPUc + (Tr(R1) + Tr(R2) + … + Tr(Rn)) * N * CPUc，此处假设HashTable构建所需CPU资源远远高于两值简单比较代价，为N * CPUc
    • IO Cost = 小表scan代价 ＋ 小表广播代价 ＋ 大表scan代价 ＝ Tr(Rsmall) * Tsz(Rsmall) * Hr + n * Tr(Rsmall) * Tsz(Rsmall) * NEt + (Tr(R1)* Tsz(R1) + … + Tr(Rn) * Tsz(Rn)) * Hr

• 选择最优执行路径(代价最小的执行路径)
  通常使用动态规划, 从各种执行路径中找出代价最小的执行路径