Analysis

从SQL或者DataFrame API中解析得到抽象语法树,依据catalog元数据校验语法树(表名、列名,或列类型),将Unresolved Logical Plan解析成Resolved Logical Plan

多个性质类似的Rule组成一个Batch,多个Batch构成一个Batchs,这些batches会由RuleExecutor执行,先按一个一个Batch顺序执行,然后对Batch里面的每个Rule顺序执行。每个Batch会执行一次会多次。

Logical Optimizations

基于规则优化,其中包含谓词下推、列裁剪、常亮折叠等。利用Rule(规则)将Resolved Logical Plan解析成Optimized Logical Plan,同样是由RuleExecutor执行

Physical Planning

前面的Logical Plan不能被Spark执行,这个过程是把Logical Plan转换成多个Physical Plan(物理计划),然后利用Cost Mode(代价模型)选择最佳的执行计划;

和前面的逻辑计划绑定和优化不一样,这里使用Strategy(策略),而前面介绍的逻辑计划绑定和优化经过transform动作之后,树的类型没有改变,也就是说:Expression 经过 transformations 之后得到的还是 Expression ;Logical Plan 经过 Transformations 之后得到的还是Logical Plan。而到了这个阶段，经过 Transformations 动作之后，树的类型改变了，由Logical Plan转换成Physical Plan了。 一个Logical Plan(逻辑计划)经过一系列的策略处理之后,得到多个物理计划,物理计划在Spark是由SparkPlan实现的。多个Physical Plan再经过Cost Model(代价模型,CBO)得到选择后的物理计划(Selected Physical Plan)

CBO

估算所有可能的物理计划的代价，并挑选出代价最小的物理执行计划。

Cost = rows * weight + size * (1 - weight)

• rows:记录行数代表了 CPU 代价
• size:代表了 IO 代价
• spark.sql.cbo.joinReorder.card.weight

LogicalPlan统计信息

LogicalPlanStats以trait的方式在每个LogicalPlan中实现

/** * A trait to add statistics propagation to [[LogicalPlan]]. */
trait LogicalPlanStats { self: LogicalPlan =>
  def stats: Statistics = statsCache.getOrElse {
    // 开启cbo 统计,只实现了Aggregate、Filter、Join、Project
    // 其余逻辑还是复用SizeInBytesOnlyStatsPlanVisitor
    // 主要统计 rowCount,size,ColumnStat(列统计信息)
    if (conf.cboEnabled) {
      // 除了统计节点的字节数
      statsCache = Option(BasicStatsPlanVisitor.visit(self))
    } else {
      // 仅仅统计节点的大小(以字节为单位)
      statsCache = Option(SizeInBytesOnlyStatsPlanVisitor.visit(self))
    }
    statsCache.get
  }
  /** A cache for the estimated statistics, such that it will only be computed once. */
  protected var statsCache: Option[Statistics] = None
}

如果开启CBO,在Optimize阶段,会通过收集的表信息对InnerJoin sql进行优化,如下图:

Code Generation

前面生成的Physical Plan还不能直接交给Spark执行,Spark最后仍然会用一些Rule对SparkPlan进行处理,如下:

QueryExecution

/** A sequence of rules that will be applied in order to the physical plan before execution. */
  protected def preparations: Seq[Rule[SparkPlan]] = Seq(
    PlanSubqueries(sparkSession),                           // 特殊子查询物理计划处理
    EnsureRequirements(sparkSession.sessionState.conf),     // 确保执行计划分区与排序的正确性
    CollapseCodegenStages(sparkSession.sessionState.conf),  // 代码生成
    ReuseExchange(sparkSession.sessionState.conf),          // 节点重用
    ReuseSubquery(sparkSession.sessionState.conf))          // 子查询重用

Join Cardinality(基数)

Inner Join : num(A IJ B) = num(A)*num(B)/max(distinct(A.k),distinct(B.k))

Left-Outer Join : num(A LOJ B) = max(num(A IJ B),num(A))

Right-Outer Join : num(A ROJ B) = max(num(A IJ B),num(B))

Full-Outer Join : num(A FOJ B) = num(A LOJ B) + num(A ROJ B) - num(A IJ B)

cost = weight * cardinality + (1.0 - weight) * size