1. 程式人生 > >第二篇:Spark SQL Catalyst源碼分析之SqlParser

第二篇:Spark SQL Catalyst源碼分析之SqlParser

end from pop tco 循環 -c font 多個 再看

/** Spark SQL源碼分析系列文章*/

Spark SQL的核心執行流程我們已經分析完畢,可以參見Spark SQL核心執行流程,下面我們來分析執行流程中各個核心組件的工作職責。

本文先從入口開始分析,即如何解析SQL文本生成邏輯計劃的,主要設計的核心組件式SqlParser是一個SQL語言的解析器,用scala實現的Parser將解析的結果封裝為Catalyst TreeNode ,關於Catalyst這個框架後續文章會介紹。

一、SQL Parser入口

Sql Parser 其實是封裝了scala.util.parsing.combinator下的諸多Parser,並結合Parser下的一些解析方法,構成了Catalyst的組件UnResolved Logical Plan。

先來看流程圖:

技術分享

一段SQL會經過SQL Parser解析生成UnResolved Logical Plan(包含UnresolvedRelation、 UnresolvedFunction、 UnresolvedAttribute)。

在源代碼裏是:

[java] view plain copy
  1. def sql(sqlText: String): SchemaRDD = new SchemaRDD(this, parseSql(sqlText))//sql("select name,value from temp_shengli") 實例化一個SchemaRDD
  2. protected[sql] def parseSql(sql: String): LogicalPlan = parser(sql) //實例化SqlParser
  3. class SqlParser extends StandardTokenParsers with PackratParsers {
  4. def apply(input: String): LogicalPlan = { //傳入sql語句調用apply方法,input參數即sql語句
  5. // Special-case out set commands since the value fields can be
  6. // complex to handle without RegexParsers. Also this approach
  7. // is clearer for the several possible cases of set commands.
  8. if (input.trim.toLowerCase.startsWith("set")) {
  9. input.trim.drop(3).split("=", 2).map(_.trim) match {
  10. case Array("") => // "set"
  11. SetCommand(None, None)
  12. case Array(key) => // "set key"
  13. SetCommand(Some(key), None)
  14. case Array(key, value) => // "set key=value"
  15. SetCommand(Some(key), Some(value))
  16. }
  17. } else {
  18. phrase(query)(new lexical.Scanner(input)) match {
  19. case Success(r, x) => r
  20. case x => sys.error(x.toString)
  21. }
  22. }
  23. }

1. 當我們調用sql("select name,value from temp_shengli")時,實際上是new了一個SchemaRDD

2. new SchemaRDD時,構造方法調用parseSql方法,parseSql方法實例化了一個SqlParser,這個Parser初始化調用其apply方法。

3. apply方法分支:

3.1 如果sql命令是set開頭的就調用SetCommand,這個類似Hive裏的參數設定,SetCommand其實是一個Catalyst裏TreeNode之LeafNode,也是繼承自LogicalPlan,關於Catalyst的TreeNode庫這個暫不詳細介紹,後面會有文章來詳細講解。

3.2 關鍵是else語句塊裏,才是SqlParser解析SQL的核心代碼:

[java] view plain copy
  1. phrase(query)(new lexical.Scanner(input)) match {
  2. case Success(r, x) => r
  3. case x => sys.error(x.toString)
  4. }

可能 phrase方法大家很陌生,不知道是幹什麽的,那麽我們首先看一下SqlParser的類圖:技術分享

SqlParser類繼承了scala內置集合Parsers,這個Parsers。我們可以看到SqlParser現在是具有了分詞的功能,也能解析combiner的語句(類似p ~> q,後面會介紹)。

Phrase方法:

[java] view plain copy
  1. /** A parser generator delimiting whole phrases (i.e. programs).
  2. *
  3. * `phrase(p)` succeeds if `p` succeeds and no input is left over after `p`.
  4. *
  5. * @param p the parser that must consume all input for the resulting parser
  6. * to succeed.
  7. * @return a parser that has the same result as `p`, but that only succeeds
  8. * if `p` consumed all the input.
  9. */
  10. def phrase[T](p: Parser[T]) = new Parser[T] {
  11. def apply(in: Input) = lastNoSuccessVar.withValue(None) {
  12. p(in) match {
  13. case s @ Success(out, in1) =>
  14. if (in1.atEnd)
  15. s
  16. else
  17. lastNoSuccessVar.value filterNot { _.next.pos < in1.pos } getOrElse Failure("end of input expected", in1)
  18. case ns => lastNoSuccessVar.value.getOrElse(ns)
  19. }
  20. }
  21. }


Phrase是一個循環讀取輸入字符的方法,如果輸入in沒有到達最後一個字符,就繼續對parser進行解析,直到最後一個輸入字符。

我們註意到Success這個類,出現在Parser裏, 在else塊裏最終返回的也有Success:

[java] view plain copy
  1. /** The success case of `ParseResult`: contains the result and the remaining input.
  2. *
  3. * @param result The parser‘s output
  4. * @param next The parser‘s remaining input
  5. */
  6. case class Success[+T](result: T, override val next: Input) extends ParseResult[T] {

通過源碼可知,Success封裝了當前解析器的解析結果result, 和還沒有解析的語句。

所以上面判斷了Success的解析結果中in1.atEnd? 如果輸入流結束了,就返回s,即Success對象,這個Success包含了SqlParser解析的輸出。

二、Sql Parser核心

在SqlParser裏phrase接受2個參數:

第一個是query,一種帶模式的解析規則,返回的是LogicalPlan。

第二個是lexical詞匯掃描輸入。

SqlParser parse的流程是,用lexical詞匯掃描接受SQL關鍵字,使用query模式來解析符合規則的SQL。

2.1 lexical keyword

在SqlParser裏定義了KeyWord這個類: [java] view plain copy
  1. protected case class Keyword(str: String)
在我使用的spark1.0.0版本裏目前只支持了一下SQL保留字: [java] view plain copy
  1. protected val ALL = Keyword("ALL")
  2. protected val AND = Keyword("AND")
  3. protected val AS = Keyword("AS")
  4. protected val ASC = Keyword("ASC")
  5. protected val APPROXIMATE = Keyword("APPROXIMATE")
  6. protected val AVG = Keyword("AVG")
  7. protected val BY = Keyword("BY")
  8. protected val CACHE = Keyword("CACHE")
  9. protected val CAST = Keyword("CAST")
  10. protected val COUNT = Keyword("COUNT")
  11. protected val DESC = Keyword("DESC")
  12. protected val DISTINCT = Keyword("DISTINCT")
  13. protected val FALSE = Keyword("FALSE")
  14. protected val FIRST = Keyword("FIRST")
  15. protected val FROM = Keyword("FROM")
  16. protected val FULL = Keyword("FULL")
  17. protected val GROUP = Keyword("GROUP")
  18. protected val HAVING = Keyword("HAVING")
  19. protected val IF = Keyword("IF")
  20. protected val IN = Keyword("IN")
  21. protected val INNER = Keyword("INNER")
  22. protected val INSERT = Keyword("INSERT")
  23. protected val INTO = Keyword("INTO")
  24. protected val IS = Keyword("IS")
  25. protected val JOIN = Keyword("JOIN")
  26. protected val LEFT = Keyword("LEFT")
  27. protected val LIMIT = Keyword("LIMIT")
  28. protected val MAX = Keyword("MAX")
  29. protected val MIN = Keyword("MIN")
  30. protected val NOT = Keyword("NOT")
  31. protected val NULL = Keyword("NULL")
  32. protected val ON = Keyword("ON")
  33. protected val OR = Keyword("OR")
  34. protected val OVERWRITE = Keyword("OVERWRITE")
  35. protected val LIKE = Keyword("LIKE")
  36. protected val RLIKE = Keyword("RLIKE")
  37. protected val UPPER = Keyword("UPPER")
  38. protected val LOWER = Keyword("LOWER")
  39. protected val REGEXP = Keyword("REGEXP")
  40. protected val ORDER = Keyword("ORDER")
  41. protected val OUTER = Keyword("OUTER")
  42. protected val RIGHT = Keyword("RIGHT")
  43. protected val SELECT = Keyword("SELECT")
  44. protected val SEMI = Keyword("SEMI")
  45. protected val STRING = Keyword("STRING")
  46. protected val SUM = Keyword("SUM")
  47. protected val TABLE = Keyword("TABLE")
  48. protected val TRUE = Keyword("TRUE")
  49. protected val UNCACHE = Keyword("UNCACHE")
  50. protected val UNION = Keyword("UNION")
  51. protected val WHERE = Keyword("WHERE")
這裏根據這些保留字,反射,生成了一個SqlLexical
[java] view plain copy
  1. override val lexical = new SqlLexical(reservedWords)
SqlLexical利用它的Scanner這個Parser來讀取輸入,傳遞給query。

2.2 query

query的定義是Parser[LogicalPlan] 和 一堆奇怪的連接符(其實都是Parser的方法啦,看上圖),*,~,^^^,看起來很讓人費解。通過查閱讀源碼,以下列出幾個常用的: | is the alternation combinator. It says “succeed if either the left or right operand parse successfully”
左邊算子和右邊的算子只要有一個成功了,就返回succeed,類似or

~ is the sequential combinator. It says “succeed if the left operand parses successfully, and then the right parses successfully on the remaining input”
左邊的算子成功後,右邊的算子對後續的輸入也計算成功,就返回succeed

opt `opt(p)` is a parser that returns `Some(x)` if `p` returns `x` and `None` if `p` fails.
如果p算子成功則返回則返回Some(x) 如果p算子失敗,返回fails

^^^ `p ^^^ v` succeeds if `p` succeeds; discards its result, and returns `v` instead.
如果左邊的算子成功,取消左邊算子的結果,返回右邊算子。

~> says “succeed if the left operand parses successfully followed by the right, but do not include the left content in the result”
如果左邊的算子和右邊的算子都成功了,返回的結果中不包含左邊的返回值。
protected lazy val limit: Parser[Expression] =
LIMIT ~> expression

<~ is the reverse, “succeed if the left operand is parsed successfully followed by the right, but do not include the right content in the result”
這個和~>操作符的意思相反,如果左邊的算子和右邊的算子都成功了,返回的結果中不包含右邊的
termExpression <~ IS ~ NOT ~ NULL ^^ { case e => IsNotNull(e) } |

^^{} 或者 ^^=> is the transformation combinator. It says “if the left operand parses successfully, transform the result using the function on the right”
rep => simply says “expect N-many repetitions of parser X” where X is the parser passed as an argument to rep
變形連接符,意思是如果左邊的算子成功了,用^^右邊的算子函數作用於返回的結果 接下來看query的定義: [java] view plain copy
  1. protected lazy val query: Parser[LogicalPlan] = (
  2. select * (
  3. UNION ~ ALL ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Union(q1, q2) } |
  4. UNION ~ opt(DISTINCT) ^^^ { (q1: LogicalPlan, q2: LogicalPlan) => Distinct(Union(q1, q2)) }
  5. )
  6. | insert | cache
  7. )
沒錯,返回的是一個Parser,裏面的類型是LogicalPlan。 query的定義其實是一種模式,用到了上述的諸多操作符,如|, ^^, ~> 等等 給定一種sql模式,如select,select xxx from yyy where ccc =ddd 如果匹配這種寫法,則返回Success,否則返回Failure. 這裏的模式是select 模式後面可以接union all 或者 union distinct。 即如下書寫式合法的,否則出錯。 [java] view plain copy
  1. select a,b from c
  2. union all
  3. select e,f from g
這個 *號是一個repeat符號,即可以支持多個union all 子句。 看來目前spark1.0.0只支持這三種模式,即select, insert, cache。

那到底是怎麽生成LogicalPlan的呢? 我們再看一個詳細的: [java] view plain copy
  1. protected lazy val select: Parser[LogicalPlan] =
  2. SELECT ~> opt(DISTINCT) ~ projections ~
  3. opt(from) ~ opt(filter) ~
  4. opt(grouping) ~
  5. opt(having) ~
  6. opt(orderBy) ~
  7. opt(limit) <~ opt(";") ^^ {
  8. case d ~ p ~ r ~ f ~ g ~ h ~ o ~ l =>
  9. val base = r.getOrElse(NoRelation)
  10. val withFilter = f.map(f => Filter(f, base)).getOrElse(base)
  11. val withProjection =
  12. g.map {g =>
  13. Aggregate(assignAliases(g), assignAliases(p), withFilter)
  14. }.getOrElse(Project(assignAliases(p), withFilter))
  15. val withDistinct = d.map(_ => Distinct(withProjection)).getOrElse(withProjection)
  16. val withHaving = h.map(h => Filter(h, withDistinct)).getOrElse(withDistinct)
  17. val withOrder = o.map(o => Sort(o, withHaving)).getOrElse(withHaving)
  18. val withLimit = l.map { l => Limit(l, withOrder) }.getOrElse(withOrder)
  19. withLimit
  20. }
這裏我給稱它為select模式。 看這個select語句支持什麽模式的寫法: select distinct projections from filter grouping having orderBy limit. 給出一個符合的該select 模式的sql, 註意到 帶opt連接符的是可選的,可以寫distinct也可以不寫。 [java] view plain copy
  1. select game_id, user_name from game_log where date<=‘2014-07-19‘ and user_name=‘shengli‘ group by game_id having game_id > 1 orderBy game_id limit 50.

projections是什麽呢? 其實是一個表達式,是一個Seq類型,一連串的表達式可以使 game_id也可以是 game_id AS gmid 。 返回的確實是一個Expression,是Catalyst裏TreeNode。 [java] view plain copy
  1. protected lazy val projections: Parser[Seq[Expression]] = repsep(projection, ",")
  2. protected lazy val projection: Parser[Expression] =
  3. expression ~ (opt(AS) ~> opt(ident)) ^^ {
  4. case e ~ None => e
  5. case e ~ Some(a) => Alias(e, a)()
  6. }

模式裏from是什麽的? 其實是一個relations,就是一個關系,在SQL裏可以是表,表join表 [java] view plain copy
  1. protected lazy val from: Parser[LogicalPlan] = FROM ~> relations
[java] view plain copy
  1. protected lazy val relation: Parser[LogicalPlan] =
  2. joinedRelation |
  3. relationFactor
  4. protected lazy val relationFactor: Parser[LogicalPlan] =
  5. ident ~ (opt(AS) ~> opt(ident)) ^^ {
  6. case tableName ~ alias => UnresolvedRelation(None, tableName, alias)
  7. } |
  8. "(" ~> query ~ ")" ~ opt(AS) ~ ident ^^ { case s ~ _ ~ _ ~ a => Subquery(a, s) }
  9. protected lazy val joinedRelation: Parser[LogicalPlan] =
  10. relationFactor ~ opt(joinType) ~ JOIN ~ relationFactor ~ opt(joinConditions) ^^ {
  11. case r1 ~ jt ~ _ ~ r2 ~ cond =>
  12. Join(r1, r2, joinType = jt.getOrElse(Inner), cond)
  13. }

這裏看出來,其實就是table之間的操作,但是返回的Subquery確實是一個LogicalPlan [java] view plain copy
  1. case class Subquery(alias: String, child: LogicalPlan) extends UnaryNode {
  2. override def output = child.output.map(_.withQualifiers(alias :: Nil))
  3. override def references = Set.empty
  4. }

scala裏的語法糖很多,這樣寫的確比較方便,但是對初學者可能有點晦澀了。 至此我們知道,SqlParser是怎麽生成LogicalPlan的了。

三、總結

本文從源代碼剖析了Spark Catalyst 是如何將Sql解析成Unresolved邏輯計劃(包含UnresolvedRelation、 UnresolvedFunction、 UnresolvedAttribute)的。 sql文本作為輸入,實例化了SqlParser,SqlParser的apply方法被調用,分別處理2種輸入,一種是命令參數,一種是sql。對應命令參數的會生成一個葉子節點,SetCommand,對於sql語句,會調用Parser的phrase方法,由lexical的Scanner來掃描輸入,分詞,最後由query這個由我們定義好的sql模式利用parser的連接符來驗證是否符合sql標準,如果符合則隨即生成LogicalPlan語法樹,不符合則會提示解析失敗。 通過對spark catalyst sql parser的解析,使我理解了,sql語言的語法標準是如何實現的和如何解析sql生成邏輯計劃語法樹。 ——EOF——

原創文章,轉載請註明:

轉載自:OopsOutOfMemory盛利的Blog,作者: OopsOutOfMemory

本文鏈接地址:http://blog.csdn.net/oopsoom/article/details/37943507

註:本文基於署名-非商業性使用-禁止演繹 2.5 中國大陸(CC BY-NC-ND 2.5 CN)協議,歡迎轉載、轉發和評論,但是請保留本文作者署名和文章鏈接。如若需要用於商業目的或者與授權方面的協商,請聯系我。

技術分享

轉自:http://blog.csdn.net/oopsoom/article/details/37943507

第二篇:Spark SQL Catalyst源碼分析之SqlParser