扩展 Spark SQL 解析
共 3727字，需浏览 8分钟
·
2021-04-24 23:58
大家好久不见了，今天跟大家分享下Spark吧，谈谈如何修改Spark SQL解析，让其更符合你的业务逻辑。好，我们开始吧...
理论基础
ANTLR
Antlr4是一款开源的语法分析器生成工具，能够根据语法规则文件生成对应的语法分析器。现在很多流行的应用和开源项目里都有使用，比如Hadoop、Hive以及Spark等都在使用ANTLR来做语法分析。
ANTLR 语法识别一般分为二个阶段：
1.词法分析阶段 （lexical analysis)
对应的分析程序叫做 lexer ，负责将符号（token）分组成符号类（token class or token type）
2.解析阶段
根据词法，构建出一棵分析树（parse tree）或叫语法树（syntax tree）
ANTLR的语法文件，非常像电路图，从入口到出口，每个Token就像电阻，连接线就是短路点。
语法文件（*.g4）
上面截图对应的语法文件片段，定义了两部分语法，一部分是显示表达式和赋值，另外一部分是运算和表达式定义。
stat: expr NEWLINE # printExpr
| ID '=' expr NEWLINE # assign
| NEWLINE # blank
;
expr: expr op=('*'|'/') expr # MulDiv
| expr op=('+'|'-') expr # AddSub
| INT # int
| ID # id
| '(' expr ')' # parens
;
接下来，加上定义词法部分，就能形成完整的语法文件。
完整语法文件：
grammar LabeledExpr; // rename to distinguish from Expr.g4
prog: stat+ ;
stat: expr NEWLINE # printExpr
| ID '=' expr NEWLINE # assign
| NEWLINE # blank
;
expr: expr op=('*'|'/') expr # MulDiv
| expr op=('+'|'-') expr # AddSub
| INT # int
| ID # id
| '(' expr ')' # parens
;
MUL : '*' ; // assigns token name to '*' used above in grammar
DIV : '/' ;
ADD : '+' ;
SUB : '-' ;
ID : [a-zA-Z]+ ; // match identifiers
INT : [0-9]+ ; // match integers
NEWLINE:'\r'? '\n' ; // return newlines to parser (is end-statement signal)
WS : [ \t]+ -> skip ; // toss out whitespace
SqlBase.g4
Spark的语法文件，在sql下的catalyst模块里，如下图：
扩展语法定义
一条正常SQL，例如 Select t.id,t.name from t , 现在我们为其添加一个 JACKY表达式，令其出现在 Select 后面 ，形成一条语句
Select t.id,t.name JACKY(2) from t
我们先看一下正常的语法规则：
现在我们添加一个 jackyExpression
jackExpression 本身的规则就是 JACKY加上括号包裹的一个数字
将 JACKY 添加为token
修改语法文件 如下：
jackyExpression
: JACKY'(' number ')'
//expression
;
namedExpression
: expression (AS? (identifier | identifierList))?
;
namedExpressionSeq
: namedExpression (',' namedExpression | jackyExpression )*
;
扩展逻辑计划
经过上面的修改，就可以测试语法规则，是不是符合预期了，下面是一颗解析树,我们可以看到jackyExpression已经可以正常解析了。
Spark 执行流程
这里引用一张经典的Spark SQL架构图
我们输入的 SQL语句 首先被解析成 Unresolved Logical Pan ，对应的是
给逻辑计划添加遍历方法：
override def visitJackyExpression(ctx: JackyExpressionContext): String = withOrigin(ctx) {
println("this is astbuilder jacky = "+ctx.number().getText)
this.jacky = ctx.number().getText.toInt
ctx.number().getText
}
再处理namedExpression的时候，添加jackyExpression处理
// Expressions.
val expressions = Option(namedExpressionSeq).toSeq
.flatMap(_.namedExpression.asScala)
.map(typedVisit[Expression])
//jackyExpression 处理
if(namedExpressionSeq().jackyExpression()!=null && namedExpressionSeq().jackyExpression().size() > 0){
visitJackyExpression(namedExpressionSeq().jackyExpression().get(0))
}
好了，到这里从逻辑计划处理就完成了，有了逻辑计划，就可以在后续物理计划中添加相应的处理逻辑就可以了（还没研究明白... Orz）。
测试
测试用例
public class Case4 {
public static void main(String[] args) {
CharStream ca = CharStreams.fromString("SELECT `b`.`id`,`b`.`class` JACKY(2) FROM `b` LIMIT 10");
SqlBaseLexer lexer = new SqlBaseLexer(ca);
SqlBaseParser sqlBaseParser = new SqlBaseParser(new CommonTokenStream(lexer));
ParseTree parseTree = sqlBaseParser.singleStatement();
AstBuilder astBuilder = new AstBuilder();
astBuilder.visit(parseTree);
System.out.println(parseTree.toStringTree(sqlBaseParser));
System.out.println(astBuilder.jacky());
}
}
执行结果
关注 【 麒思妙想】解锁更多硬核。
历史文章导读：
如果文章对您有那么一点点帮助，我将倍感荣幸
欢迎 关注、在看、点赞、转发