Spark SQL原理解析前言: "Spark SQL源碼剖析(一)SQL解析框架Catalyst流程概述" 這一次要開始真正介紹Spark解析SQL的流程,首先是從Sql Parse階段開始,簡單點說,這個階段就是使用Antlr4,將一條Sql語句解析成語法樹。 可能有童鞋沒接觸過antlr4這個 ...
Spark SQL原理解析前言:
Spark SQL源碼剖析(一)SQL解析框架Catalyst流程概述
這一次要開始真正介紹Spark解析SQL的流程,首先是從Sql Parse階段開始,簡單點說,這個階段就是使用Antlr4,將一條Sql語句解析成語法樹。
可能有童鞋沒接觸過antlr4這個內容,推薦看看《antlr4權威指南》前四章,看完起碼知道antlr4能幹嘛。我這裡就不多介紹了。
這篇首先先介紹調用spark.sql()時候的流程,再看看antlr4在這個其中的主要功能,最後再將探究Logical Plan究竟是什麼東西。
初始流程
當你調用spark.sql的時候,會調用下麵的方法:
def sql(sqlText: String): DataFrame = {
Dataset.ofRows(self, sessionState.sqlParser.parsePlan(sqlText))
}
parse sql階段主要是parsePlan(sqlText)這一部分。而這裡又會輾轉去org.apache.spark.sql.catalyst.parser.AbstractSqlParser調用parse方法。這裡貼下關鍵代碼。
protected def parse[T](command: String)(toResult: SqlBaseParser => T): T = {
logDebug(s"Parsing command: $command")
val lexer = new SqlBaseLexer(new UpperCaseCharStream(CharStreams.fromString(command)))
lexer.removeErrorListeners()
lexer.addErrorListener(ParseErrorListener)
lexer.legacy_setops_precedence_enbled = SQLConf.get.setOpsPrecedenceEnforced
val tokenStream = new CommonTokenStream(lexer)
val parser = new SqlBaseParser(tokenStream)
parser.addParseListener(PostProcessor)
parser.removeErrorListeners()
parser.addErrorListener(ParseErrorListener)
parser.legacy_setops_precedence_enbled = SQLConf.get.setOpsPrecedenceEnforced
try {
try {
// first, try parsing with potentially faster SLL mode
parser.getInterpreter.setPredictionMode(PredictionMode.SLL)
toResult(parser)
}
catch {
case e: ParseCancellationException =>
// if we fail, parse with LL mode
tokenStream.seek(0) // rewind input stream
parser.reset()
// Try Again.
parser.getInterpreter.setPredictionMode(PredictionMode.LL)
toResult(parser)
}
}
catch {
case e: ParseException if e.command.isDefined =>
throw e
case e: ParseException =>
throw e.withCommand(command)
case e: AnalysisException =>
val position = Origin(e.line, e.startPosition)
throw new ParseException(Option(command), e.message, position, position)
}
}
可以發現,這裡面的處理邏輯,無論是SqlBaseLexer還是SqlBaseParser都是Antlr4的東西,包括最後的toResult(parser)也是調用訪問者模式的類去遍歷語法樹來生成Logical Plan。如果對antlr4有一定瞭解,那麼對這裡這些東西一定不會陌生。那我們接下來看看Antlr4在這其中的角色。
Antlr4生成語法樹
Spark提供了一個.g4文件,編譯的時候會使用Antlr根據這個.g4生成對應的詞法分析類和語法分析類,同時還使用了訪問者模式,用以構建Logical Plan(語法樹)。
訪問者模式簡單說就是會去遍歷生成的語法樹(針對語法樹中每個節點生成一個visit方法),以及返回相應的值。我們接下來看看一條簡單的select語句生成的樹是什麼樣子。
這個sqlBase.g4文件我們也可以直接拿出來玩,直接複製出來,用antlr相關工具就可以生成一個生成一個解析SQL的圖了。
這裡antlr4和grun都已經存儲成bat文件,所以可以直接調用,實際命令在《antlr4權威指南》說得很詳細了就不介紹了。調用完後就會生成這樣的語法樹。
這裡,將SELECT TABLE_A.B FROM TABLE_A,轉換成一棵語法樹。我們可以看到這顆語法樹非常複雜,這是因為SQL解析中,要適配這種SELECT語句之外,還有很多其他類型的語句,比如INSERT,ALERT等等。Spark SQL這個模塊的最終目標,就是將這樣的一棵語法樹轉換成一個可執行的Dataframe(RDD)。
我們現階段的目標則是要先生成Logical Plan,Spark使用Antlr4的訪問者模式,生成Logical Plan。這裡順便說下怎麼實現訪問者模式吧,在使用antlr4命令的時候,加上-visit參數就會生成SqlBaseBaseVisitor,裡面提供了預設的訪問各個節點的觸發方法。我們可以通過繼承這個類,重寫對應節點的visit方法,實現自己的訪問邏輯,而這個繼承的類就是org.apache.spark.sql.catalyst.parser.AstBuilder。
通過觀察這棵樹,我們可以發現針對我們的SELECT語句,比較重要的一個節點,是querySpecification節點,實際上,在AstBuilder類中,visitQuerySpecification也是比較重要的一個方法(訪問對應節點時觸發),正是在這個方法中生成主要的Logical Plan的。
接下來重點看這個方法,以及探究Logical Plan。
生成Logical Plan
我們先看看AstBuilder中的代碼:
class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging {
......其他代碼
override def visitQuerySpecification(
ctx: QuerySpecificationContext): LogicalPlan = withOrigin(ctx) {
val from = OneRowRelation().optional(ctx.fromClause) { //如果有FROM語句,生成對應的Logical Plan
visitFromClause(ctx.fromClause)
}
withQuerySpecification(ctx, from)
}
......其他代碼
代碼中會先判斷是否有FROM子語句,有的話會去生成對應的Logical Plan,再調用withQuerySpecification()方法,而withQuerySpecification()方法是比較核心的一個方法。它會處理包括SELECT,FILTER,GROUP BY,HAVING等子語句的邏輯。
代碼比較長就不貼了,有興趣的童鞋可以去看看,大意就是使用scala的模式匹配,匹配不同的子語句生成不同的Logical Plan。
然後再來說說最終生成的LogicalPlan,LogicalPlan其實是繼承自TreeNode,所以本質上LogicalPlan就是一棵樹。
而實際上,LogicalPlan還有多個子類,分別表示不同的SQL子語句。
- LeafNode,葉子節點,一般用來表示用戶命令
- UnaryNode,一元節點,表示FILTER等操作
- BinaryNode,二元節點,表示JOIN,GROUP BY等操作
這裡一元二元這些都是對應關係代數方面的知識,在學資料庫理論的時候肯定有接觸過,不過估計都還給老師了吧(/偷笑)。不過一元二元基本上也就是用來區分具體的操作,如上面說的FILTER,或是JOIN等,也不是很複雜。這三個類都位於org.apache.spark.sql.catalyst.plans.logical.LogicalPlan中,有興趣的童鞋可以看看。而後,這三個類又會有多個子類,用以表示不同的情況,這裡就不再贅述。
最後看看用一個測試案例,看看會生成什麼吧。示例中簡單生成一個臨時的view,然後直接select查詢這個view。代碼如下:
//生成DataFrame
val df = Seq((1, 1)).toDF("key", "value")
df.createOrReplaceTempView("src")
//調用spark.sql
val queryCaseWhen = sql("select key from src ")
補充下,這裡的sql()方法是做了一些封裝的方法,可以直接看成spark.sql(...)。最終經過parse SQL後會變成如下的內容:
'Project ['key]
+- 'UnresolvedRelation `src`
這個Project是UnaryNode的一個子類(SELECT自然是一元節點),表明我們要查詢的欄位是key。
UnresolvedRelation是一個新的概念,這裡順便說下,我們通過SQL parse生成的這棵樹,其實叫Unresolved LogicalPlan,這裡的Unresolved的意思說,還不知道src是否存在,或它的元數據是什麼樣,只有通過Analysis階段後,才會把Unresolved變成Resolved LogicalPlan。這裡的意思可以理解為,讀取名為src的表,但這張表的情況未知,有待驗證。
總的來說,我們的示例足夠簡單直接,所以內容會比較少,不過拿來學習是足夠了。
下一個階段是要使用這棵樹進行分析驗證了,也就是Analysis階段,這一塊留到下篇介紹吧。
以上~