用户
 找回密码
 立即注册

QQ登录

只需一步,快速开始

搜索
查看: 1789|回复: 0

第四篇 Spark SQL Catalyst源码分析之TreeNode Library

[复制链接]

394

主题

412

帖子

2065

积分

管理员

Rank: 9Rank: 9Rank: 9

积分
2065

活跃会员热心会员推广达人宣传达人灌水之王突出贡献优秀版主荣誉管理论坛元老

发表于 2018-1-6 21:02:20 | 显示全部楼层 |阅读模式
----------------------------------------------------------
Spark SQL源码分析系列文章
    前几篇文章介绍了Spark SQL的Catalyst的核心运行流程SqlParser,和Analyzer,本来打算直接写Optimizer的,但是发现忘记介绍TreeNode这个Catalyst的核心概念,介绍这个可以更好的理解Optimizer是如何对Analyzed Logical Plan进行优化的生成Optimized Logical Plan,本文就将TreeNode基本架构进行解释。


一、TreeNode类型   TreeNode Library是Catalyst的核心类库,语法树的构建都是由一个个TreeNode组成。TreeNode本身是一个BaseType <: TreeNode[BaseType] 的类型,并且实现了Product这个trait,这样可以存放异构的元素了。
   TreeNode有三种形态:BinaryNodeUnaryNodeLeaf Node.
   在Catalyst里,这些Node都是继承自Logical Plan,可以说每一个TreeNode节点就是一个Logical Plan(包含Expression)(直接继承自TreeNode)
   主要继承关系类图如下:
1.png
1、BinaryNode
二元节点,即有左右孩子的二叉节点
[Java] 纯文本查看 复制代码
[[TreeNode]] that has two children, [[left]] and [[right]].  
trait BinaryNode[BaseType <: TreeNode[BaseType]] {  
  def left: BaseType  
  def right: BaseType  
  def children = Seq(left, right)  
}  
abstract class BinaryNode extends LogicalPlan with trees.BinaryNode[LogicalPlan] {  
  self: Product =>  
}  

节点定义比较简单,左孩子,右孩子都是BaseType。 children是一个Seq(left, right)
下面列出主要继承二元节点的类,可以当查询手册用 :)
这里提示下平常常用的二元节点:Join和Union

2.png
2、UnaryNode
一元节点,即只有一个孩子节点
[Java] 纯文本查看 复制代码
 A [[TreeNode]] with a single [[child]].  
trait UnaryNode[BaseType <: TreeNode[BaseType]] {  
  def child: BaseType  
  def children = child :: Nil  
}  
abstract class UnaryNode extends LogicalPlan with trees.UnaryNode[LogicalPlan] {  
  self: Product =>  
}  

下面列出主要继承一元节点的类,可以当查询手册用 :)
常用的二元节点有,Project,Subquery,Filter,Limit ...等
3.png
3、Leaf Node
叶子节点,没有孩子节点的节点。
[Java] 纯文本查看 复制代码
A [[TreeNode]] with no children.  
trait LeafNode[BaseType <: TreeNode[BaseType]] {  
  def children = Nil  
}  
abstract class LeafNode extends LogicalPlan with trees.LeafNode[LogicalPlan] {  
  self: Product =>  
  // Leaf nodes by definition cannot reference any input attributes.  
  override def references = Set.empty  
}  

下面列出主要继承叶子节点的类,可以当查询手册用 :)
提示常用的叶子节点: Command类系列,一些Funtion函数,以及Unresolved Relation...etc.
4.png
二、TreeNode 核心方法  简单介绍一个TreeNode这个类的属性和方法
  currentId
  一颗树里的TreeNode有个唯一的id,类型是java.util.concurrent.atomic.AtomicLong原子类型。
[Java] 纯文本查看 复制代码
private val currentId = new java.util.concurrent.atomic.AtomicLong  
protected def nextId() = currentId.getAndIncrement()  

  sameInstance
  判断2个实例是否是同一个的时候,只需要判断TreeNode的id。
[Java] 纯文本查看 复制代码
def sameInstance(other: TreeNode[_]): Boolean = {  
  this.id == other.id  
}  


  fastEquals,更常用的一个快捷的判定方法,没有重写Object.Equals,这样防止scala编译器生成case class equals 方法
[Java] 纯文本查看 复制代码
def fastEquals(other: TreeNode[_]): Boolean = {  
   sameInstance(other) || this == other  
 }  


  map,flatMap,collect都是递归的对子节点进行应用PartialFunction,其它方法还有很多,篇幅有限这里不一一描述了。


2.1、核心方法 transform 方法  transform该方法接受一个PartialFunction,就是就是前一篇文章Analyzer里提到的Batch里面的Rule。
  是会将Rule迭代应用到该节点的所有子节点,最后返回这个节点的副本(一个和当前节点不同的节点,后面会介绍,其实就是利用反射来返回一个修改后的节点)。
  如果rule没有对一个节点进行PartialFunction的操作,就返回这个节点本身。

  来看一个例子:
[Java] 纯文本查看 复制代码
object GlobalAggregates extends Rule[LogicalPlan] {  
  def apply(plan: LogicalPlan): LogicalPlan = plan transform {   //apply方法这里调用了logical plan(TreeNode) 的transform方法来应用一个PartialFunction。  
    case Project(projectList, child) if containsAggregates(projectList) =>  
      Aggregate(Nil, projectList, child)  
  }  
  def containsAggregates(exprs: Seq[Expression]): Boolean = {  
    exprs.foreach(_.foreach {  
      case agg: AggregateExpression => return true  
      case _ =>  
    })  
    false  
  }  
}  

这个方法真正的调用是transformChildrenDown,这里提到了用先序遍历来对子节点进行递归的Rule应用。
如果在对当前节点应用rule成功,修改后的节点afterRule,来对其children节点进行rule的应用。
transformDown方法:
[Java] 纯文本查看 复制代码
/** 
* Returns a copy of this node where `rule` has been recursively applied to it and all of its 
* children (pre-order). When `rule` does not apply to a given node it is left unchanged. 
* @param rule the function used to transform this nodes children 
*/  
ef transformDown(rule: PartialFunction[BaseType, BaseType]): BaseType = {  
 val afterRule = rule.applyOrElse(this, identity[BaseType])  
 // Check if unchanged and then possibly return old copy to avoid gc churn.  
 if (this fastEquals afterRule) {  
   transformChildrenDown(rule)  //修改前节点this.transformChildrenDown(rule)  
 } else {  
   afterRule.transformChildrenDown(rule) //修改后节点进行transformChildrenDown  
 }  

  最重要的方法transformChildrenDown:
  对children节点进行递归的调用PartialFunction,利用最终返回的newArgs来生成一个新的节点,这里调用了makeCopy()来生成节点。
transformChildrenDown方法:
[Java] 纯文本查看 复制代码
/** 
 * Returns a copy of this node where `rule` has been recursively applied to all the children of 
 * this node.  When `rule` does not apply to a given node it is left unchanged. 
 * @param rule the function used to transform this nodes children 
 */  
def transformChildrenDown(rule: PartialFunction[BaseType, BaseType]): this.type = {  
  var changed = false  
  val newArgs = productIterator.map {  
    case arg: TreeNode[_] if children contains arg =>  
      val newChild = arg.asInstanceOf[BaseType].transformDown(rule) //递归子节点应用rule  
      if (!(newChild fastEquals arg)) {  
        changed = true  
        newChild  
      } else {  
        arg  
      }  
    case Some(arg: TreeNode[_]) if children contains arg =>  
      val newChild = arg.asInstanceOf[BaseType].transformDown(rule)  
      if (!(newChild fastEquals arg)) {  
        changed = true  
        Some(newChild)  
      } else {  
        Some(arg)  
      }  
    case m: Map[_,_] => m  
    case args: Traversable[_] => args.map {  
      case arg: TreeNode[_] if children contains arg =>  
        val newChild = arg.asInstanceOf[BaseType].transformDown(rule)  
        if (!(newChild fastEquals arg)) {  
          changed = true  
          newChild  
        } else {  
          arg  
        }  
      case other => other  
    }  
    case nonChild: AnyRef => nonChild  
    case null => null  
  }.toArray  
  if (changed) makeCopy(newArgs) else this //根据作用结果返回的newArgs数组,反射生成新的节点副本。  
}  

  makeCopy方法,反射生成节点副本  
[Java] 纯文本查看 复制代码
/** 
  * Creates a copy of this type of tree node after a transformation. 
  * Must be overridden by child classes that have constructor arguments 
  * that are not present in the productIterator. 
  * @param newArgs the new product arguments. 
  */  
 def makeCopy(newArgs: Array[AnyRef]): this.type = attachTree(this, "makeCopy") {  
   try {  
     val defaultCtor = getClass.getConstructors.head  //反射获取默认构造函数的第一个  
     if (otherCopyArgs.isEmpty) {  
       defaultCtor.newInstance(newArgs: _*).asInstanceOf[this.type] //反射生成当前节点类型的节点  
     } else {  
       defaultCtor.newInstance((newArgs ++ otherCopyArgs).toArray: _*).asInstanceOf[this.type] //如果还有其它参数,++  
     }  
   } catch {  
     case e: java.lang.IllegalArgumentException =>  
       throw new TreeNodeException(  
         this, s"Failed to copy node.  Is otherCopyArgs specified correctly for $nodeName? "  
           + s"Exception message: ${e.getMessage}.")  
   }  
 }

  
三、TreeNode实例  现在准备从一段sql来出发,画一下这个spark sql的整体树的transformation。
SELECT * FROM (SELECT * FROM src) a join (select * from src)b on a.key=b.key
首先,我们先执行一下,在控制台里看一下生成的计划:
[Java] 纯文本查看 复制代码
<span style="font-size:12px;">sbt/sbt hive/console  
Using /usr/java/default as default JAVA_HOME.  
Note, this will be overridden by -java-home if it is set.  
[info] Loading project definition from /app/hadoop/shengli/spark/project/project  
[info] Loading project definition from /app/hadoop/shengli/spark/project  
[info] Set current project to root (in build file:/app/hadoop/shengli/spark/)  
[info] Starting scala interpreter...  
[info]   
import org.apache.spark.sql.catalyst.analysis._  
import org.apache.spark.sql.catalyst.dsl._  
import org.apache.spark.sql.catalyst.errors._  
import org.apache.spark.sql.catalyst.expressions._  
import org.apache.spark.sql.catalyst.plans.logical._  
import org.apache.spark.sql.catalyst.rules._  
import org.apache.spark.sql.catalyst.types._  
import org.apache.spark.sql.catalyst.util._  
import org.apache.spark.sql.execution  
import org.apache.spark.sql.hive._  
import org.apache.spark.sql.hive.test.TestHive._  
import org.apache.spark.sql.parquet.ParquetTestData  
    
scala> val query = sql("SELECT * FROM (SELECT * FROM src) a join (select * from src)b on a.key=b.key")</span>  



3.1、UnResolve Logical Plan  第一步生成UnResolve Logical Plan 如下:
[Java] 纯文本查看 复制代码
scala> query.queryExecution.logical  
res0: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan =   
Project  
 Join Inner, Some(('a.key = 'b.key))  
  Subquery a  
   Project  
    UnresolvedRelation None, src, None  
  Subquery b  
   Project  
    UnresolvedRelation None, src, None  


  如果画成树是这样的,仅个人理解:
  我将一开始介绍的三种Node分别用绿色UnaryNode,红色Binary Node 和 蓝色 LeafNode 来表示。
5.png

3.2、Analyzed Logical Plan
  Analyzer会将允用Batch的Rules来对Unresolved Logical  Plan Tree 进行rule应用,这里用来EliminateAnalysisOperators将Subquery给消除掉,Batch("Resolution将Atrribute和Relation给Resolve了,Analyzed Logical Plan Tree如下图:
6.png
3.3、Optimized Plan  我把Catalyst里的Optimizer戏称为Spark SQL的优化大师,因为整个Spark SQL的优化都是在这里进行的,后面会有文章来讲解Optimizer。
  在这里,优化的不明显,因为SQL本身不复杂
[Java] 纯文本查看 复制代码
scala> query.queryExecution.optimizedPlan  
res3: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan =   
Project [key#0,value#1,key#2,value#3]  
 Join Inner, Some((key#0 = key#2))  
  MetastoreRelation default, src, None  
  MetastoreRelation default, src, None  


生成的树如下图:

7.png
3.4、executedPlan  最后一步是最终生成的物理执行计划,里面涉及到了Hive的TableScan,涉及到了HashJoin操作,还涉及到了Exchange,Exchange涉及到了Shuffle和Partition操作。
[Java] 纯文本查看 复制代码
scala> query.queryExecution.executedPlan  
res4: org.apache.spark.sql.execution.SparkPlan =   
Project [key#0:0,value#1:1,key#2:2,value#3:3]  
 HashJoin [key#0], [key#2], BuildRight  
  Exchange (HashPartitioning [key#0:0], 150)  
   HiveTableScan [key#0,value#1], (MetastoreRelation default, src, None), None  
  Exchange (HashPartitioning [key#2:0], 150)  
   HiveTableScan [key#2,value#3], (MetastoreRelation default, src, None), None


生成的物理执行树如图:
8.png

四、总结:    本文介绍了Spark SQL的Catalyst框架核心TreeNode类库,绘制了TreeNode继承关系的类图,了解了TreeNode这个类在Catalyst所起到的作用。语法树中的Logical Plan均派生自TreeNode,并且Logical Plan派生出TreeNode的三种形态,即Binary Node, Unary Node, Leaft Node。 正式这几种节点,组成了Spark SQl的Catalyst的语法树。
  TreeNode的transform方法是核心的方法,它接受一个rule,会对当前节点的孩子节点进行递归的调用rule,最后会返回一个TreeNode的copy,这种操作就是transformation,贯穿了Spark SQL执行的几个核心阶段,如Analyze,Optimize阶段。
  最后用一个实际的例子,展示出来Spark SQL的执行树生成流程。
  
  我目前的理解就是这些,如果分析不到位的地方,请大家多多指正。

——EOF——
原创文章,转载请注明:



回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

站长推荐 上一条 /4 下一条

返回顶部