Spark Sql 执行流程源码阅读笔记

2021-04-23

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住给大家分享一下。点击跳转到网站：https://www.captainai.net/dongkelun

欢迎关注我的公众号

前言

学习Spark SQL解析sql生成DataFrame的源码，进行记录

基于Spark3.0.1

代码

以下面的代码作为入口，进行跟踪调试

1 2	val df = spark.sql("select * from test_parquet") df.show

主要的类、方法

spark.sql即SparkSession.sql

/**
 * Executes a SQL query using Spark, returning the result as a `DataFrame`.
 * The dialect that is used for SQL parsing can be configured with 'spark.sql.dialect'.
 *
 * @since 2.0.0
 */
def sql(sqlText: String): DataFrame = withActive {
  val tracker = new QueryPlanningTracker
  val plan = tracker.measurePhase(QueryPlanningTracker.PARSING) {
    sessionState.sqlParser.parsePlan(sqlText)
  }
  Dataset.ofRows(self, plan, tracker)
}

sessionState.sqlParser.parsePlan(sqlText)

sessionState.sqlParser val sqlParser: ParserInterface
这里的sqlParser是SparkSqlParser，

为什么是SparkSqlParser,在类BaseSessionStateBuilder里:
详细的流程后面单独写
1
2
3
protected lazy val sqlParser: ParserInterface = {
   extensions.buildParser(session, new SparkSqlParser(conf))
 }

/**
 * Concrete parser for Spark SQL statements.
 */
class SparkSqlParser(conf: SQLConf) extends AbstractSqlParser(conf) {
  val astBuilder = new SparkSqlAstBuilder(conf)

  private val substitutor = new VariableSubstitution(conf)

  protected override def parse[T](command: String)(toResult: SqlBaseParser => T): T = {
    super.parse(substitutor.substitute(command))(toResult)
  }
}

abstract class AbstractSqlParser(conf: SQLConf) extends ParserInterface with Logging {
    ...

SparkSqlParser没有重写AbstractSqlParser的parsePlan方法，即sessionState.sqlParser.parsePlan实际上调用的是AbstractSqlParser的parsePlan方法

/** Creates LogicalPlan for a given SQL string. */
override def parsePlan(sqlText: String): LogicalPlan = parse(sqlText) { parser =>
  astBuilder.visitSingleStatement(parser.singleStatement()) match {
    case plan: LogicalPlan => plan
    case _ =>
      val position = Origin(None, None)
      throw new ParseException(Option(sqlText), "Unsupported SQL statement", position, position)
  }
}

parse

这里的parse方法为SparkSqlParser的parse

1
2
3

protected override def parse[T](command: String)(toResult: SqlBaseParser => T): T = {
  super.parse(substitutor.substitute(command))(toResult)
}

接着调用父类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)

    val tokenStream = new CommonTokenStream(lexer)
    val parser = new SqlBaseParser(tokenStream)
    parser.addParseListener(PostProcessor)
    parser.removeErrorListeners()
    parser.addErrorListener(ParseErrorListener)
    parser.legacy_setops_precedence_enbled = conf.setOpsPrecedenceEnforced
    parser.legacy_exponent_literal_as_decimal_enabled = conf.exponentLiteralAsDecimalEnabled
    parser.SQL_standard_keyword_behavior = conf.ansiEnabled

    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)
    }
  }
}

先执行parse方法再执行toResult(parser),其实就是parsePlan的方法体

{ parser =>
    astBuilder.visitSingleStatement(parser.singleStatement()) match {
      case plan: LogicalPlan => plan
      case _ =>
        val position = Origin(None, None)
        throw new ParseException(Option(sqlText), "Unsupported SQL statement", position, position)
    }
  }

astBuilder

本文由 董可伦 发表于伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。

非商业转载请注明作者及出处。商业转载请联系作者本人。

本文标题：Spark Sql 执行流程源码阅读笔记

本文链接：https://dongkelun.com/2021/04/23/sparkSqlExecutionFlow/

欢迎关注我的公众号