[大数据]Spark SQL实现原理-逻辑计划的优化-操作下推：PushProjectionThroughUnion

PushProjectionThroughUnion操作下推优化规则的作用是：把在Union操作一边的Projections（投影）操作推到Union的两边。要注意这样优化的前提是在Spark SQL中nion操作不会对数据去重。这里的Projections可以理解为select字段的操作。也就是说，把select操作推到Union操作的两边。

优化规则的使用

先通过例子来查看和理解一下该优化规则的效果。使用local模式启动spark的scala终端，打开TRACE日志（可以在scala终端中通过sc.setLogLevel(“TRACE”)来开启跟踪日志）。在终端中输入以下代码：

import spark.implicits._
case class Person(name: String, age: Long)

val data2 = Seq(Person("Michael", 29), Person("Andy", 30), Person("Justin", 11))
val ds2 = spark.createDataset(data2)
val data3 = Seq(Person("Michael", 9), Person("Andy", 7), Person("Justin", 11))
val ds3 = spark.createDataset(data3)

// 进行数据集的union和select运算
ds2.union(ds3).union(ds2).select("name").explain(true)

查看输出的日志，可以看到以下输出：

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PushProjectionThroughUnion ===
!Project [name#119]                         Union
!+- Union                                   :- Project [name#119]
!   :- LocalRelation [name#119, age#120L]   :  +- LocalRelation [name#119, age#120L]
!   :- LocalRelation [name#124, age#125L]   :- Project [name#124]
!   +- LocalRelation [name#119, age#120L]   :  +- LocalRelation [name#124, age#125L]
!                                           +- Project [name#119]
!                                              +- LocalRelation [name#119, age#120L]

我们做了两次union操作和一次select操作。从以上日志输出可以看到，初始的逻辑计划中Project操作只是在Union的一边，通过优化器的PushProjectionThroughUnion规则优化后，Project操作被放到了Union的两边。

那么，为什么要这样去优化呢？在查询数据时，要让查询的性能提升，就要尽量少扫描数据块，通过union下推的优化，可以让本来只有union一边的Project或Filter操作放到了Union的两边，这样做其实是把本来只过滤一个Dataset的操作，优化成了同时过滤两个Dataset，从而对两边的数据进行了过滤，让计算的数据量变得更少了，从而提升了union操作的性能。

另外，由于Union操作不会对数据进行去重，所以这种优化不会对数据产生丢失。Union distinct操作却不能这样优化。

规则的实现

Union的Project下推操作的详细实现可以在PushProjectionThroughUnion类中查看。主要逻辑的其实现代码在apply函数中，该函数的实现逻辑如下：

（1）检查逻辑计划的节点，若父节点是Project，子节点是Union操作时（也就是先进行union操作，再进行select操作时，该优化规则只处理这种逻辑计划），跳到（2），否则什么都不做。

（2）从判断条件可知，头节点是一个Project节点，我们需要把Project推到Union操作的两端，就必须要Union作为父节点，所以要重新创建头节点（Project节点）；

（3）然后为其他子节点创建Project父节点；

（4）最后把这颗子树，作为Union的子树，让Union作为父节点。

其实现代码如下：

// 后续遍历逻辑计划树，查找父节点是Project，而子节点是Union操作的
def apply(plan: LogicalPlan): LogicalPlan = plan transform {

  // 查找父节点是Project，而子节点是Union操作的节点
  case p @ Project(projectList, Union(children)) =>
    assert(children.nonEmpty)
    if (projectList.forall(_.deterministic)) {  // 遍历所有的字段名
      // 从新创建Union的子节点，头节点是一个Project节点
      val newFirstChild = Project(projectList, children.head)
      // 构建其他Project节点和其子节点
      val newOtherChildren = children.tail.map { child =>
        val rewrites = buildRewrites(children.head, child)
        Project(projectList.map(pushToRight(_, rewrites)), child)
      }
      // 以Union为父节点，把两边的Project节点都作为子节点
      Union(newFirstChild +: newOtherChildren)
    } else {
      p
    }
}

注意：以上只是简单的介绍了函数的实现逻辑，要了解详细的实现，可以直接查看实现源码。

小结

本文介绍了逻辑计划优化规则的操作下推类优化规则中的一个：PushProjectionThroughUnion的使用和实现原理。该优化规则主要优化union和select组合使用的情况。为了提升计算性能，Spark SQL会把select操作推到Union操作的两端，这样可以减少进行union操作时的数据量。

针对union或join等操作后续还有很多优化规则，都是基于这样的考虑。

这也提醒我们，在进行数据处理时，随时提醒自己 ：过滤！过滤！过滤！先去掉不需要的行和列。