Spark SQL 处理流程分析（一）

下面的代码演示了通过Case Class进行表Schema定义的例子：

// sc is an existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
// this is used to implicitly convert an RDD to a DataFrame.
import sqlContext.implicits._// Define the schema using a case class.
// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,
// you can use custom classes that implement the Product interface.
case class Person(name: String, age: Int)// Create an RDD of Person objects and register it as a table.
val people = sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
people.registerTempTable("people")// SQL statements can be run by using the sql methods provided by sqlContext.
val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")// The results of SQL queries are DataFrames and support all the normal RDD operations.
// The columns of a row in the result can be accessed by field index:
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)// or by field name:
teenagers.map(t => "Name: " + t.getAs[String]("name")).collect().foreach(println)// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]
teenagers.map(_.getValuesMap[Any](List("name", "age"))).collect().foreach(println)
// Map("name" -> "Justin", "age" -> 19)

（1）sql方法返回DataFrame

  def sql(sqlText: String): DataFrame = {DataFrame(this, parseSql(sqlText))}

其中parseSql(sqlText)方法生成相应的LogicalPlan得到，该方法源码如下：

//根据传入的sql语句，生成LogicalPlan
protected[sql] def parseSql(sql: String): LogicalPlan = ddlParser.parse(sql, false)

ddlParser对象定义如下：

protected[sql] val sqlParser = new SparkSQLParser(getSQLDialect().parse(_))
protected[sql] val ddlParser = new DDLParser(sqlParser.parse(_))

（2）然后调用DataFrame的apply方法

private[sql] object DataFrame {def apply(sqlContext: SQLContext, logicalPlan: LogicalPlan): DataFrame = {new DataFrame(sqlContext, logicalPlan)}
}

可以看到，apply方法参数有两个，分别是SQLContext和LogicalPlan，调用的是DataFrame的构造方法，具体源码如下：

//DataFrame构造方法，该构造方法会自动对LogicalPlan进行分析，然后返回QueryExecution对象
def this(sqlContext: SQLContext, logicalPlan: LogicalPlan) = {this(sqlContext, {val qe = sqlContext.executePlan(logicalPlan)//判断是否已经创建，如果是则抛异常if (sqlContext.conf.dataFrameEagerAnalysis) {qe.assertAnalyzed()  // This should force analysis and throw errors if there are any}qe})}

（3）val qe = sqlContext.executePlan(logicalPlan) 返回QueryExecution， sqlContext.executePlan方法源码如下：

protected[sql] def executePlan(plan: LogicalPlan) =new sparkexecution.QueryExecution(this, plan)

QueryExecution类中表达了Spark执行SQL的主要工作流程，具体如下

class QueryExecution(val sqlContext: SQLContext, val logical: LogicalPlan) {@VisibleForTestingdef assertAnalyzed(): Unit = sqlContext.analyzer.checkAnalysis(analyzed)lazy val analyzed: LogicalPlan = sqlContext.analyzer.execute(logical)lazy val withCachedData: LogicalPlan = {assertAnalyzed()sqlContext.cacheManager.useCachedData(analyzed)}lazy val optimizedPlan: LogicalPlan = sqlContext.optimizer.execute(withCachedData)// TODO: Don't just pick the first one...lazy val sparkPlan: SparkPlan = {SparkPlan.currentContext.set(sqlContext)sqlContext.planner.plan(optimizedPlan).next()}// executedPlan should not be used to initialize any SparkPlan. It should be// only used for execution.lazy val executedPlan: SparkPlan = sqlContext.prepareForExecution.execute(sparkPlan)/** Internal version of the RDD. Avoids copies and has no schema *///调用toRDD方法执行任务将结果转换为RDDlazy val toRdd: RDD[InternalRow] = executedPlan.execute()protected def stringOrError[A](f: => A): String =try f.toString catch { case e: Throwable => e.toString }def simpleString: String = {s"""== Physical Plan ==|${stringOrError(executedPlan)}""".stripMargin.trim}override def toString: String = {def output =analyzed.output.map(o => s"${o.name}: ${o.dataType.simpleString}").mkString(", ")s"""== Parsed Logical Plan ==|${stringOrError(logical)}|== Analyzed Logical Plan ==|${stringOrError(output)}|${stringOrError(analyzed)}|== Optimized Logical Plan ==|${stringOrError(optimizedPlan)}|== Physical Plan ==|${stringOrError(executedPlan)}|Code Generation: ${stringOrError(executedPlan.codegenEnabled)}""".stripMargin.trim}
}

可以看到，SQL的执行流程为
1.Parsed Logical Plan：LogicalPlan
2.Analyzed Logical Plan：
lazy val analyzed: LogicalPlan = sqlContext.analyzer.execute(logical)
3.Optimized Logical Plan：lazy val optimizedPlan: LogicalPlan = sqlContext.optimizer.execute(withCachedData)
4. Physical Plan：lazy val executedPlan: SparkPlan = sqlContext.prepareForExecution.execute(sparkPlan)

可以调用results.queryExecution方法查看，代码如下：

scala> results.queryExecution
res1: org.apache.spark.sql.SQLContext#QueryExecution =
== Parsed Logical Plan ==
'Project [unresolvedalias('name)]
 'UnresolvedRelation [people], None== Analyzed Logical Plan ==
name: string
Project [name#0]
 Subquery people
  LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at createDataFrame at <console>:47== Optimized Logical Plan ==
Project [name#0]
 LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at createDataFrame at <console>:47== Physical Plan ==
TungstenProject [name#0]
 Scan PhysicalRDD[name#0,age#1]Code Generation: true

（4）然后调用DataFrame的主构造器完成DataFrame的构造

class DataFrame private[sql](@transient val sqlContext: SQLContext,@DeveloperApi @transient val queryExecution: QueryExecution) extends Serializable

（5）
当调用DataFrame的collect等方法时，便会触发执行executedPlan

  def collect(): Array[Row] = withNewExecutionId {queryExecution.executedPlan.executeCollect()}

例如：

scala> results.collect
res6: Array[org.apache.spark.sql.Row] = Array([Michael], [Andy], [Justin])

整体流程图如下：

Spark SQL 处理流程分析（一）相关推荐

Spark SQL运行流程及性能优化：RBO和CBO
1 Spark SQL运行流程 1.1 Spark SQL核心--Catalyst Spark SQL的核心是Catalyst查询编译器,它将用户程序中的SQL/Dataset/DataFrame经过 ...
Spark SQL Catalyst源代码分析之TreeNode Library
/** Spark SQL源代码分析系列文章*/ 前几篇文章介绍了Spark SQL的Catalyst的核心执行流程.SqlParser,和Analyzer,本来打算直接写Optimizer的,可是发 ...
Spark SQL Catalyst源代码分析Optimizer
/** Spark SQL源代码分析系列*/ 前几篇文章介绍了Spark SQL的Catalyst的核心运行流程.SqlParser,和Analyzer 以及核心类库TreeNode,本文将具体解说S ...
【华为云技术分享】大数据实践解析（下）：Spark的读写流程分析
摘要:本文通过简单的例子来解析,大数据实践中的Spark读写流程,内容主要聚焦于Spark中的高效并行读写以及在写过程中如何保证事务性. 导读: 众所周知,在大数据/数据库领域,数据的存储格式直接影响 ...
MyBatis系列之浅谈SQL执行流程分析
目录独立使用Mybatis Mybatis执行流程 SqlSessionFactory\SqlSession MapperProxy Excutor 独立使用Mybatis 这篇文章主要以分析Myb ...
TP5 框架 SQL 执行流程分析及 5.0.9 SQL 注入漏洞分析
文章目录 SQL查询流程 TP 5.0.9 SQL注入修复 SQL查询流程 TP5手册:https://www.kancloud.cn/manual/thinkphp5/118044 在分析 tp5 ...
Spark SQL执行流程解析之 sql执行顺序
一.Spark SQL模板解析 Spark Sql 关键字执行顺序跟Sql执行顺序类似: 1.先将from前两表做笛卡尔积加载进来形成虚拟表vt1, 2.on条件后为真的插入虚拟表形成vt2, 如果包 ...
深入浅出Mybatis系列（十）---SQL执行流程分析（源码篇）
原文地址:http://www.cnblogs.com/dongying/p/4142476.html 最近太忙了,一直没时间继续更新博客,今天忙里偷闲继续我的Mybatis学习之旅.在前九篇中,介绍 ...
MySQL 文件结构、逻辑架构及 sql 执行流程分析作者：Java后端架构
1.MySQL 文件说明 1.1 MySQL 文件夹文件 linux 服务器上 MySQL 安装好之后都有如下文件: auto.cnf:每一个 MySQL 实例都有一个唯一 ID 蓝色文件夹:表示数据 ...

Spark SQL 处理流程分析（一）

Spark SQL 处理流程分析（一）相关推荐

最新文章

热门文章

Spark SQL 处理流程分析 （一）

Spark SQL 处理流程分析 （一）相关推荐

最新文章

热门文章

Spark SQL 处理流程分析（一）

Spark SQL 处理流程分析（一）相关推荐