Spark SQL源码分析之核心流程

标签：spark sql   spark   hadoop   hive   分布式计算   

    前一段时间测试过Shark，并且对Spark SQL也进行了一些测试，但是还是忍不住对Spark SQL一探究竟，就从源代码的角度来看一下Spark SQL的核心执行流程吧。

一、引子

1. val sqlContext = new org.apache.spark.sql.SQLContext(sc)
2. import sqlContext._
3.case class Person(name: String, age: Int)
4.val people = sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt))
5.people.registerAsTable("people")
6.val teenagers = sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
7.teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

 二、SQLCOntext

Catalog  

 一个存储<tableName,logicalPlan>的map结构，查找关系的目录，注册表，注销表，查询表和逻辑计划关系的类。

SqlParser 

 Parse 传入的sql来对语法分词，构建语法树，返回一个logical plan

Analyzer 

  logical plan的语法分析器

Optimizer 

 logical Plan的优化器

LogicalPlan 

逻辑计划，由catalyst的TreeNode组成，可以看到有3种语法树

SparkPlanner 

包含不同策略的优化策略来优化物理执行计划

QueryExecution 

sql执行的环境上下文

就是这些对象组成了Spark SQL的运行时，看起来很酷，有静态的metadata存储，有分析器、优化器、逻辑计划、物理计划、执行运行时。
那这些对象是怎么相互协作来执行sql语句的呢？

三、Spark SQL执行流程

      

核心组件都是绿色的方框，每一步流程的结果都是蓝色的框框，调用的方法是橙色的框框。

先概括一下，大致的执行流程是：
Parse SQL -> Analyze Logical Plan -> Optimize Logical Plan -> Generate Physical Plan -> Prepare Spark Plan -> Executre SQL -> Generate RDD

更具体的执行流程：

     sql or hql -> sql parser(parse)生成 unresolved logical plan -> analyzer(analysis)生成analyzed logical plan  -> optimizer(optimize)optimized logical plan -> spark planner(use stretage to plan)生成physical plan -> call next函数生成spark plan -> spark plan(prepare) prepared spark plan -> call toRDD 执行sql生成RDD

3.1、Parse SQL

	  /**
   * Executes a SQL query using Spark, returning the result as a SchemaRDD.
   *
   * @group userf
   */
   def sql(sqlText: String): SchemaRDD = new SchemaRDD(this, parseSql(sqlText))

   @transient
  protected[sql] val parser = new catalyst.SqlParser  


  protected[sql] def parseSql(sql: String): LogicalPlan = parser(sql)

 3.2、Analyze to Execution

 override def collect(): Array[Row] = queryExecution.executedPlan.executeCollect()

protected abstract class QueryExecution {
    def logical: LogicalPlan

    lazy val analyzed = analyzer(logical)  //首先分析器会分析逻辑计划
    lazy val optimizedPlan = optimizer(analyzed) //随后优化器去优化分析后的逻辑计划
    // TODO: Don't just pick the first one...
    lazy val sparkPlan = planner(optimizedPlan).next() //根据策略生成plan物理计划
    // executedPlan should not be used to initialize any SparkPlan. It should be
    // only used for execution.
    lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan) //最后生成已经准备好的Spark Plan

    /** Internal version of the RDD. Avoids copies and has no schema */
    lazy val toRdd: RDD[Row] = executedPlan.execute() //最后调用toRDD方法执行任务将结果转换为RDD

    protected def stringOrError[A](f: => A): String =
      try f.toString catch { case e: Throwable => e.toString }

    def simpleString: String = stringOrError(executedPlan)

    override def toString: String =
      s"""== Logical Plan ==
         |${stringOrError(analyzed)}
         |== Optimized Logical Plan ==
         |${stringOrError(optimizedPlan)}
         |== Physical Plan ==
         |${stringOrError(executedPlan)}
      """.stripMargin.trim
  }

  四、总结：

原创文章：转载请注明出自：http://blog.csdn.net/oopsoom/article/details/37658021

Spark SQL源码分析之核心流程,布布扣,bubuko.com

Spark SQL源码分析之核心流程

标签：spark sql   spark   hadoop   hive   分布式计算   

原文地址：http://blog.csdn.net/oopsoom/article/details/37658021