Spark 源碼系列（九）Spark SQL 初體驗之解析過程詳解

時間  2019-11-07
標籤 spark 源碼 系列 sql 體驗 解析 過程 詳解 欄目 Spark 简体版
原文   原文鏈接

很久沒更新博客了，以前學了一些 R 語言和機器學習的內容，作了一些筆記，以後也會放到博客上面來給你們共享。一個月前就打算更新 Spark Sql 的內容了，由於一些別的事情耽誤了，今天就簡單寫點，Spark1.2 立刻就要出來了，不知道變更會不會很大，聽說添加了不少的新功能呢，期待中...html

首先聲明一下這個版本的代碼是 1.1 的，以前講的都是 1.0 的。sql

Spark 支持兩種模式，一種是在 spark 裏面直接寫 sql，能夠經過 sql 來查詢對象，相似. net 的 LINQ 同樣，另一種支持 hive 的 HQL。不論是哪一種方式，下面提到的步驟都會有，不一樣的是具體的執行過程。下面就說一下這個過程。數據庫

Sql 解析成 LogicPlanapache

使用 Idea 的快捷鍵 Ctrl + Shift + N 打開 SQLQuerySuite 文件，進行調試吧。api

def sql(sqlText: String): SchemaRDD = {
    if (dialect == "sql") {
      new SchemaRDD(this, parseSql(sqlText))
    } else {
      sys.error(s"Unsupported SQL dialect: $dialect")
    }
  }
複製代碼

從這裏能夠看出來，第一步是解析 sql，最後把它轉換成一個 SchemaRDD。點擊進入 parseSql 函數，發現解析 Sql 的過程在 SqlParser 這個類裏面。 在 SqlParser 的 apply 方法裏面，咱們能夠看到 else 語句裏面的這段代碼。緩存

//對input進行解析，符合query的模式的就返回Success
      phrase(query)(new lexical.Scanner(input)) match {
        case Success(r, x) => r
        case x => sys.error(x.toString)
      }
複製代碼

這裏咱們主要關注 query 就能夠。app

protected lazy val query: Parser[LogicalPlan] = (
    select * (
        UNION ~ ALL ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Union(q1, q2) } |
        INTERSECT ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Intersect(q1, q2) } |
        EXCEPT ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Except(q1, q2)} |
        UNION ~ opt(DISTINCT) ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Distinct(Union(q1, q2)) }
      )
    | insert | cache
  )
複製代碼

這裏面有不少看不懂的操做符，請到下面這個網址裏面去學習。這裏能夠看出來它目前支持的 sql 語句只是 select 和 insert。機器學習

www.scala-lang.org/api/2.10.4/…ide

咱們繼續查看 select。函數

// ~>只保留右邊的模式 opt可選的 ~按順序合成 <~只保留左邊的
  protected lazy val select: Parser[LogicalPlan] =
    SELECT ~> opt(DISTINCT) ~ projections ~
    opt(from) ~ opt(filter) ~
    opt(grouping) ~
    opt(having) ~
    opt(orderBy) ~
    opt(limit) <~ opt(";") ^^ {
      case d ~ p ~ r ~ f ~ g ~ h ~ o ~ l  =>
        val base = r.getOrElse(NoRelation)
        val withFilter = f.map(f => Filter(f, base)).getOrElse(base)
        val withProjection =
          g.map {g =>
            Aggregate(assignAliases(g), assignAliases(p), withFilter)
          }.getOrElse(Project(assignAliases(p), withFilter))
        val withDistinct = d.map(_ => Distinct(withProjection)).getOrElse(withProjection)
        val withHaving = h.map(h => Filter(h, withDistinct)).getOrElse(withDistinct)
        val withOrder = o.map(o => Sort(o, withHaving)).getOrElse(withHaving)
        val withLimit = l.map { l => Limit(l, withOrder) }.getOrElse(withOrder)
        withLimit
  }
複製代碼

能夠看得出來它對 sql 的解析是和咱們經常使用的 sql 寫法是一致的，這裏面再深刻下去還有遞歸，並非看起來那麼好理解。這裏就不繼續講下去了，在解析 hive 的時候我會重點講一下，我認爲目前你們使用得更可能是仍然是來源於 hive 的數據集，畢竟 hive 那麼穩定。

到這裏咱們能夠知道第一步是經過 Parser 把 sql 解析成一個 LogicPlan。

LogicPlan 到 RDD 的轉換過程

好，下面咱們回到剛纔的代碼，接着咱們應該看 SchemaRDD。

override def compute(split: Partition, context: TaskContext): Iterator[Row] =
    firstParent[Row].compute(split, context).map(_.copy())

  override def getPartitions: Array[Partition] = firstParent[Row].partitions

  override protected def getDependencies: Seq[Dependency[_]] =
    List(new OneToOneDependency(queryExecution.toRdd))
複製代碼

SchemaRDD 是一個 RDD 的話，那麼它最重要的 3 個屬性：compute 函數，分區，依賴全在這裏面，其它的函數咱們就不看了。

挺奇怪的是，咱們 new 出來的 RDD，怎麼會有依賴呢，這個 queryExecution 是啥，點擊進去看看吧，代碼跳轉到 SchemaRDD 繼承的 SchemaRDDLike 裏面。

lazy val queryExecution = sqlContext.executePlan(baseLogicalPlan)
protected[sql] def executePlan(plan: LogicalPlan): this.QueryExecution =
    new this.QueryExecution { val logical = plan }
複製代碼

把這兩段很短的代碼都放一塊兒了，executePlan 方法就是 new 了一個 QueryExecution 出來，那咱們繼續看看 QueryExecution 這個類吧。

lazy val analyzed = ExtractPythonUdfs(analyzer(logical))
    lazy val optimizedPlan = optimizer(analyzed)
    lazy val sparkPlan = {
      SparkPlan.currentContext.set(self)
      planner(optimizedPlan).next()
    }
    // 在須要的時候加入Shuffle操做
    lazy val executedPlan: SparkPlan = prepareForExecution(sparkPlan)
    lazy val toRdd: RDD[Row] = executedPlan.execute()
複製代碼

從這裏能夠看出來 LogicPlan 是通過了 5 個步驟的轉換，要被 analyzer 和 optimizer 的處理，而後轉換成 SparkPlan，在執行以前還要被 prepareForExecution 處理一下，最後調用 execute 方法轉成 RDD.

下面咱們分步講這些個東東究竟是幹啥了。

首先咱們看看 Anayzer，它是繼承自 RuleExecutor 的，這裏插句題外話，Spark sql 的做者 Michael Armbrust 在 2013 年的 Spark Submit 上介紹 Catalyst 的時候，就說到要從總體地去優化一個 sql 的執行是很困難的，全部設計成這種基於一個一個小規則的這種優化方式，既簡單又方便維護。

好，咱們接下來看看 RuleExecutor 的 apply 方法。

def apply(plan: TreeType): TreeType = {
    var curPlan = plan
    //規則還分批次的，分批對plan進行處理
    batches.foreach { batch =>
      val batchStartPlan = curPlan
      var iteration = 1
      var lastPlan = curPlan
      var continue = true

      // Run until fix point (or the max number of iterations as specified in the strategy.
      while (continue) {
        //用batch種的小規則從左到右挨個對plan進行處理
        curPlan = batch.rules.foldLeft(curPlan) {
          case (plan, rule) =>
            val result = rule(plan)
            result
        }
        iteration += 1
        //超過了規定的迭代次數就要退出的
        if (iteration > batch.strategy.maxIterations) {
              continue = false
        }
        //通過處理成功的plan是會發生改變的，若是和上一次處理接觸的plan同樣，這說明已經沒有優化空間了，能夠結束，這個就是前面提到的Fixed point
        if (curPlan.fastEquals(lastPlan)) {
          continue = false
        }
        lastPlan = curPlan
      }
    }

    curPlan
  }
複製代碼

看完了 RuleExecutor，咱們繼續看 Analyzer，下面我只貼出來 batches 這塊的代碼，剩下的要本身去看了哦。

val batches: Seq[Batch] = Seq(
    //碰到繼承自MultiInstanceRelations接口的LogicPlan時，發現id之後重複的，就強制要求它們生成一個新的全局惟一的id
    //涉及到InMemoryRelation、LogicRegion、ParquetRelation、SparkLogicPlan
    Batch("MultiInstanceRelations", Once,
      NewRelationInstances),
    //若是大小寫不敏感就把屬性都變成小寫
    Batch("CaseInsensitiveAttributeReferences", Once,
      (if (caseSensitive) Nil else LowercaseAttributeReferences :: Nil) : _*),
    //這個牛逼啊，竟然想迭代100次的。
    Batch("Resolution", fixedPoint,
      //解析從子節點的操做生成的屬性，通常是別名引發的，好比a.id
      ResolveReferences ::
      //經過catalog解析表名
      ResolveRelations ::
      //在select語言裏，order by的屬性每每在前面沒寫，查詢的時候也須要把這些字段查出來，排序完畢以後再刪除
      ResolveSortReferences ::
      //前面講過了
      NewRelationInstances ::
      //清除被誤認爲別名的屬性，好比sum(score) as a，其實它應該是sum(score)纔對
      //它被解析的時候解析成Project(Seq(Alias(g: Generator, _))，直接返回Generator就能夠了
      ImplicitGenerate ::
      //處理語句中的*，好比select *, count(*)
      StarExpansion ::
      //解析函數
      ResolveFunctions ::
      //解析全局的聚合函數，好比select sum(score) from table
      GlobalAggregates ::
      //解析having子句後面的聚合過濾條件，好比having sum(score) > 400
      UnresolvedHavingClauseAttributes ::
      //typeCoercionRules是hive的類型轉換規則
      typeCoercionRules :_*),
    //檢查全部節點的屬性是否都已經處理完畢了，若是還有沒解析出來的屬性，這裏就會報錯！
    Batch("Check Analysis", Once,
      CheckResolution),
    //清除多餘的操做符，如今是Subquery和LowerCaseSchema，
    //第一個是子查詢，第二個HiveContext查詢樹裏面把子節點所有轉換成小寫
    Batch("AnalysisOperators", fixedPoint,
      EliminateAnalysisOperators)
  )
複製代碼

能夠看得出來 Analyzer 是把 Unresolved 的 LogicPlan 解析成 resolved 的，解析裏面的表名、字段、函數、別名什麼的。

咱們接着看 Optimizer, 從單詞上看它是用來作優化的，可是從代碼上來看它更多的是爲了過濾咱們寫的一些垃圾語句，並無作什麼實際的優化。

object Optimizer extends RuleExecutor[LogicalPlan] {
  val batches =
      //遞歸合併相鄰的兩個limit
    Batch("Combine Limits", FixedPoint(100),
      CombineLimits) ::
    Batch("ConstantFolding", FixedPoint(100),
      //替換null值
      NullPropagation,
      //替換一些簡單的常量表達式，好比 1 in (1,2) 直接返回一個true就能夠了
      ConstantFolding,
      //簡化like語句，避免全表掃描，目前支持'%demo%', '%demo','demo*','demo'
      LikeSimplification,
      //簡化過濾條件，好比true and score > 0 直接替換成score > 0
      BooleanSimplification,
      //簡化filter，好比where 1=1 或者where 1=2，前者直接去掉這個過濾，後者這個查詢就不必作了
      SimplifyFilters,
      //簡化轉換，好比兩個比較字段的數據類型是同樣的，就不須要轉換了
      SimplifyCasts,
      //簡化大小寫轉換，好比Upper(Upper('a'))轉爲認爲是Upper('a')
      SimplifyCaseConversionExpressions) ::
    Batch("Filter Pushdown", FixedPoint(100),
      //遞歸合併相鄰的兩個過濾條件
      CombineFilters,
      //把從表達式裏面的過濾替換成，先作過濾再取表達式，而且掉過濾裏面的別名屬性
      //典型的例子 select * from (select a,b from table) where a=1
      //替換成select * from (select a,b from table where a=1)
      PushPredicateThroughProject,
      //把join的on條件中能夠在原表當中作過濾的先作過濾
      //好比select a,b from x join y on x.id = y.id and x.a >0 and y.b >0
      //這個語句能夠改寫爲 select a,b from x where x.a > 0 join (select * from y where y.b >0) on x.id = y.id
      PushPredicateThroughJoin,
      //去掉一些用不上的列
      ColumnPruning) :: Nil
}
複製代碼

真是用心良苦啊，看來咱們寫 sql 的時候仍是要注意一點的，你看人家花多大的功夫來優化咱們的爛 sql。要是我確定不優化。寫得爛就慢去吧！

接下來，就改看這一句了 planner(optimizedPlan).next() 咱們先看看 SparkPlanner 吧。

protected[sql] class SparkPlanner extends SparkStrategies {
    val sparkContext: SparkContext = self.sparkContext

    val sqlContext: SQLContext = self

    def codegenEnabled = self.codegenEnabled

    def numPartitions = self.numShufflePartitions
    //把LogicPlan轉換成實際的操做，具體操做類在org.apache.spark.sql.execution包下面
    val strategies: Seq[Strategy] =
      //把cache、set、expain命令轉化爲實際的Command
      CommandStrategy(self) ::
      //把limit轉換成TakeOrdered操做
      TakeOrdered ::
      //名字有點蠱惑人，就是轉換聚合操做
      HashAggregation ::
      //left semi join只顯示鏈接條件成立的時候鏈接左邊的表的信息
      //好比select * from table1 left semi join table2 on(table1.student_no=table2.student_no);
      //它只顯示table1中student_no在表二當中的信息，它能夠用來替換exist語句
      LeftSemiJoin ::
      //等值鏈接操做，有些優化的內容，若是表的大小小於spark.sql.autoBroadcastJoinThreshold設置的字節
      //就自動轉換爲BroadcastHashJoin，即把表緩存，相似hive的map join（順序是先判斷右表再判斷右表）。
      //這個參數的默認值是10000
      //另外作內鏈接的時候還會判斷左表右表的大小，shuffle取數據大表不動，從小表拉取數據過來計算
      HashJoin ::
      //在內存裏面執行select語句進行過濾,會作緩存
      InMemoryScans ::
      //和parquet相關的操做
      ParquetOperations ::
      //基本的操做
      BasicOperators ::
      //沒有條件的鏈接或者內鏈接作笛卡爾積
      CartesianProduct ::
      //把NestedLoop鏈接進行廣播鏈接
      BroadcastNestedLoopJoin :: Nil
      ......  
}
複製代碼

這一步是把邏輯計劃轉換成物理計劃，或者說是執行計劃了，裏面有不少概念是我之前沒聽過的，網上查了一下才知道，原來數據庫的執行計劃還有那麼多的說法，這一塊須要是專門研究數據庫的人比較瞭解了。剩下的兩步就是 prepareForExecution 和 execute 操做。

prepareForExecution 操做是檢查物理計劃當中的 Distribution 是否知足 Partitioning 的要求，若是不知足的話，須要從新弄作分區，添加 shuffle 操做，這塊暫時沒咋看懂，之後還須要仔細研究。最後調用 SparkPlan 的 execute 方法，這裏面稍微講講這塊的樹型結構。

img

sql 解析出來就是一個二叉樹的結構，不論是邏輯計劃仍是物理計劃，都是這種結構，因此在代碼裏面能夠看到 LogicPlan 和 SparkPlan 的具體實現類都是有繼承上面圖中的三種類型的節點的。

非 LeafNode 的 SparkPlan 的 execute 方法都會有這麼一句 child.execute()，由於它須要先執行子節點的 execute 來返回數據，執行的過程是一個先序遍歷。

最後把這個過程也用一個圖來表示吧，方便記憶。

img

(1) 經過一個 Parser 來把 sql 語句轉換成 Unresolved LogicPlan，目前有兩種 Parser，SqlParser 和 HiveQl。

(2) 經過 Analyzer 把 LogicPlan 當中的 Unresolved 的內容給解析成 resolved 的，這裏麪包括表名、函數、字段、別名等。

(3) 經過 Optimizer 過濾掉一些垃圾的 sql 語句。

(4) 經過 Strategies 把邏輯計劃轉換成能夠具體執行的物理計劃，具體的類有 SparkStrategies 和 HiveStrategies。

(5) 在執行前用 prepareForExecution 方法先檢查一下。

(6) 先序遍歷，調用執行計劃樹的 execute 方法。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程