Spark SQL 工作流程源码解析（四）optimization 阶段（基于 Spark 3.3.0）
2024-05-19 05:08:47
前言
本文隶属于专栏《大数据技术体系》，该专栏为笔者原创，引用请注明来源，不足和错误之处请在评论区帮忙指出，谢谢！
本专栏目录结构和参考文献请见大数据技术体系
目录
Spark SQL 工作流程源码解析（一）总览（基于 Spark 3.3.0）
Spark SQL 工作流程源码解析（二）parsing 阶段（基于 Spark 3.3.0）
Spark SQL 工作流程源码解析（三）analysis 阶段（基于 Spark 3.3.0）
Spark SQL 工作流程源码解析（四）optimization 阶段（基于 Spark 3.3.0）
Spark SQL 工作流程源码解析（五）planning 阶段（基于 Spark 3.3.0）
关联
ASOF JOIN 是什么？pandas 的 merge_asof() 如何使用？
Spark SQL 如何自定义扩展？
Spark SQL 支持哪些类型的 JOIN ？
Spark 中 Dataset.show 如何使用？有哪些值得注意的地方？
一篇文章了解 SQL 中的 CTE
一篇文章搞懂 Spark 3.x 的 CacheManager
我们要做什么？
经过一开始的parsing阶段，我们的 SQL语句/DataFrame/Dataset 变成了UnresolvedLogicalPlan。这个过程最终生成的逻辑计划实际上只和一开始我们提供的SQL语句/DataFrame/Dataset 有关。
然后，经过了analysis阶段，我们得到了AnalyzedLogicalPlan。在这个过程中，我们会根据当前的数据源环境信息，应用各种解析规则，将逻辑计划树中的各种未识别的节点替换成我们解析过的节点，然后我们会对解析后的逻辑计划树进行校验，找出其中逻辑错误的地方。
而optimization阶段，作为逻辑计划的最后一个阶段，要做什么呢？
AnalyzedLogicalPlan基本上就是根据UnresolvedLogicalPlan一一转换来的，正常来讲，此时的逻辑计划已经完全可以执行了，不影响最终的正确性。但是实际的应用中，不同开发者写的 SQL 语句可能存在着很大的差异，性能也会完全不同，此时，我们应该要想办法来消除不必要的差异，这样能充分地利用缓存，同时我们可以基于各种 SQL 优化的手段，对Analyzed Logical Plan进行处理，得到更好的更方便执行的逻辑计划树。
Spark SQL 中经过
analysis阶段的逻辑计划，并不会完全保证其正确性，具体的请参考下面讲解的Finish Analysis规则批。
我们有了什么？
在analysis阶段之后，我们得到了AnalyzedLogicalPlan，我们先看看它长什么样子：
找到入口
show()
Spark SQL 的 optimization 阶段实际上发生在 show() 方法里面。
spark.sql("SELECT * FROM t_user").show
也就是说，对于 SELECT 从句，sql() 方法只会触发 parsing 和 analysis 阶段。
def show(): Unit = show(20)
show() 默认只会展示前 20 行数据。
def show(numRows: Int): Unit = show(numRows, truncate = true)
show() 方法默认会对返回的结果进行截取。
def show(numRows: Int, truncate: Boolean): Unit = if (truncate) {println(showString(numRows, truncate = 20))} else {println(showString(numRows, truncate = 0))}
可以看到，show() 底层就是通过 println() 方法打印数据行。
关于
Dataset.show方法的使用详情请参考我的这篇博客——Spark 中 Dataset.show 如何使用？有哪些值得注意的地方？
showString
我们需要打印的数据来自 Dataset.showString，我们看看它是怎么实现的。
/*** 组合表示输出行的字符串 * @param _numRows–要显示的行数 * @param truncate–如果设置为大于0，则截断字符串以截断字符，所有单元格都将对齐。 * @param vertical–如果设置为true，则垂直打印输出行（每列值一行）。*/
private[sql] def showString(_numRows: Int,truncate: Int = 20,vertical: Boolean = false): String = {val numRows = _numRows.max(0).min(ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH - 1)// 获取 Seq[Seq[String]] 类型的数据行，如果有更多的话可能获得不只一行的数据。val tmpRows = getRows(numRows, truncate)val hasMoreData = tmpRows.length - 1 > numRowsval rows = tmpRows.take(numRows + 1)val sb = new StringBuilderval numCols = schema.fieldNames.length// 设置最小字段宽度为 3val minimumColWidth = 3if (!vertical) {// 初始化每列的宽度为最小值val colWidths = Array.fill(numCols)(minimumColWidth)// 计算每列的宽度for (row <- rows) {for ((cell, i) <- row.zipWithIndex) {colWidths(i) = math.max(colWidths(i), Utils.stringHalfWidth(cell))}}val paddedRows = rows.map { row =>row.zipWithIndex.map { case (cell, i) =>if (truncate > 0) {StringUtils.leftPad(cell, colWidths(i) - Utils.stringHalfWidth(cell) + cell.length)} else {StringUtils.rightPad(cell, colWidths(i) - Utils.stringHalfWidth(cell) + cell.length)}}}// 创建分隔行val sep: String = colWidths.map("-" * _).addString(sb, "+", "+", "+\n").toString()// 列名paddedRows.head.addString(sb, "|", "|", "|\n")sb.append(sep)// 数据paddedRows.tail.foreach(_.addString(sb, "|", "|", "|\n"))sb.append(sep)} else {// Extended display mode enabledval fieldNames = rows.headval dataRows = rows.tail// 计算字段名称和数据列的宽度val fieldNameColWidth = fieldNames.foldLeft(minimumColWidth) { case (curMax, fieldName) =>math.max(curMax, Utils.stringHalfWidth(fieldName))}val dataColWidth = dataRows.foldLeft(minimumColWidth) { case (curMax, row) =>math.max(curMax, row.map(cell => Utils.stringHalfWidth(cell)).max)}dataRows.zipWithIndex.foreach { case (row, i) =>// 长度 "+ 5" 意味着除了补齐的名称和数据以外只有一个字节长度val rowHeader = StringUtils.rightPad(s"-RECORD $i", fieldNameColWidth + dataColWidth + 5, "-")sb.append(rowHeader).append("\n")row.zipWithIndex.map { case (cell, j) =>val fieldName = StringUtils.rightPad(fieldNames(j),fieldNameColWidth - Utils.stringHalfWidth(fieldNames(j)) + fieldNames(j).length)val data = StringUtils.rightPad(cell,dataColWidth - Utils.stringHalfWidth(cell) + cell.length)s" $fieldName | $data "}.addString(sb, "", "\n", "\n")}}// 打印脚注信息if (vertical && rows.tail.isEmpty) {// 在垂直模式下，显著打印一个空的 row setsb.append("(0 rows)\n")} else if (hasMoreData) {// 对于那些记录超过 "numRows" 的数据val rowsString = if (numRows == 1) "row" else "rows"sb.append(s"only showing top $numRows $rowsString\n")}sb.toString()}
从上面的源码注释中可以看到，其他地方都是一些格式化打印的工作，真正的数据获取发生在这一行：
val tmpRows = getRows(numRows, truncate)
getRows
我们看看 Dataset.getRows 是如何实现的？
/*** 获取按特定截断和垂直要求顺序表示的行。 * @param numRows–要返回的行数 * @param truncate–如果设置为大于0，则截断字符串以截断字符，所有单元格都将对齐*/
private[sql] def getRows(numRows: Int,truncate: Int): Seq[Seq[String]] = {val newDf = toDF()val castCols = newDf.logicalPlan.output.map { col =>// 由于顶层 schema 字段 Binary 类型有特殊的打印方式，因此这里不需要将它们转换成 String 类型if (col.dataType == BinaryType) {Column(col)} else {// 非 Binary 类型的字段都会得到 Cast 类型Column(col).cast(StringType)}}val data = newDf.select(castCols: _*).take(numRows + 1)// 对于数组值，使用方括号来替代 Seq 和 Array，对于长度超过 truncate 的字符串，使用 truncate-3 的字符串和 ... 来代替schema.fieldNames.map(SchemaUtils.escapeMetaCharacters).toSeq +: data.map { row =>row.toSeq.map { cell =>val str = cell match {case null => "null"case binary: Array[Byte] => binary.map("%02X".format(_)).mkString("[", " ", "]")case _ =>// 转义元字符以不破坏 showString 格式SchemaUtils.escapeMetaCharacters(cell.toString)}if (truncate > 0 && str.length > truncate) {// 对于少于4个字符的字符串，不要显示省略号。if (truncate < 4) str.substring(0, truncate)else str.substring(0, truncate - 3) + "..."} else {str}}: Seq[String]}}
很明显，数据来源于这一行：
val data = newDf.select(castCols: _*).take(numRows + 1)
这一行涉及 2 个关键的操作：select 和 take。
select
def select(cols: Column*): DataFrame = withPlan {val untypedCols = cols.map {case typedCol: TypedColumn[_, _] =>// 检查是否 `TypedColumn` 已经被`withInputType` 插入了具体的输入类型和 schemaval needInputType = typedCol.expr.find {case ta: TypedAggregateExpression if ta.inputDeserializer.isEmpty => truecase _ => false}.isDefinedif (!needInputType) {typedCol} else {throw QueryCompilationErrors.cannotPassTypedColumnInUntypedSelectError(typedCol.toString)}case other => other}Project(untypedCols.map(_.named), logicalPlan)}
由上面getRows的源码可知，在我们的例子中，cols中每个字段的类型都是 Cast，非TypedColumn，故偏函数走的都是case other => other这个分支，即不做任何处理。
Dataset.select 所做的事情就是使用 Project 将我们之前得到的AnalyzedLogicalPlan 进行封装。除此之外，它会针对每个字段调用Column.named方法，我们看看这个方法干了什么~
Column.named
private[sql] def named: NamedExpression = expr match {case expr: NamedExpression => expr// 将未别名化的生成器保留为空的名称列表，因为 analyzer 将会在嵌套表达式的类型被解析后，显示正确的默认值。case g: Generator => MultiAlias(g, Nil)// 如果我们有一个顶层的 Cast，这是一个机会给它一个更好的别名，如果在这个 Cast 下有一个命名表达式的话。case c: Cast =>c.transformUp {case c @ Cast(_: NamedExpression, _, _, _) => UnresolvedAlias(c)} match {case ne: NamedExpression => necase _ => UnresolvedAlias(expr, Some(Column.generateAlias))}case expr: Expression => UnresolvedAlias(expr, Some(Column.generateAlias))}
由于我们例子当中每个字段都是 Cast类型，故偏函数走的是 case c: Cast => 这个分支，这里面会使用UnresolvedAlias进行封装。
UnresolvedAlias也是一个命名表达式（NamedExpression）。
变化
逻辑计划变化前：
Project [addr#8, age#9L, name#10, sex#11]
+- SubqueryAlias t_user+- View (`t_user`, [addr#8,age#9L,name#10,sex#11])+- Relation [addr#8,age#9L,name#10,sex#11] json
逻辑计划变化后：
'Project [unresolvedalias(cast(addr#8 as string), None), unresolvedalias(cast(age#9L as string), None), unresolvedalias(cast(name#10 as string), None), unresolvedalias(cast(sex#11 as string), None)]
+- Project [addr#8, age#9L, name#10, sex#11]+- SubqueryAlias t_user+- View (`t_user`, [addr#8,age#9L,name#10,sex#11])+- Relation [addr#8,age#9L,name#10,sex#11] json
可以看到，Project 里面封装了UnresolvedAlias，UnresolvedAlias里面封装了Cast，Cast 里面封装了Column，完全符合上面的源码解析。
withPlan
上面生成的UnresolvedAlias不应该出现在optimization阶段，故我们会再一次触发analysis阶段。
我们注意到 select 方法里面调用了 Dataset.withPlan。
/*** 一个方便的函数，用于包装逻辑计划并生成 DataFrame。*/@inline private def withPlan(logicalPlan: LogicalPlan): DataFrame = {Dataset.ofRows(sparkSession, logicalPlan)}
如果看了我的上一讲博客的同学，相信对Dataset.ofRows感到不陌生。
def ofRows(sparkSession: SparkSession, logicalPlan: LogicalPlan): DataFrame =sparkSession.withActive {val qe = sparkSession.sessionState.executePlan(logicalPlan)qe.assertAnalyzed()new Dataset[Row](qe, RowEncoder(qe.analyzed.schema))}
具体的源码解析不再赘述，建议同学再回头看看上一讲——Spark SQL 工作流程源码解析（三）analysis 阶段（基于 Spark 3.3.0）
此处我们贴一下配置了spark.sql.planChangeLog.level的控制台打印：
22/04/01 22:28:41 INFO PlanChangeLogger:
=== Applying Rule org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveAliases ===
!'Project [unresolvedalias(cast(addr#8 as string), None), unresolvedalias(cast(age#9L as string), None), unresolvedalias(cast(name#10 as string), None), unresolvedalias(cast(sex#11 as string), None)] Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11] +- Project [addr#8, age#9L, name#10, sex#11]+- SubqueryAlias t_user +- SubqueryAlias t_user+- View (`t_user`, [addr#8,age#9L,name#10,sex#11]) +- View (`t_user`, [addr#8,age#9L,name#10,sex#11])+- Relation [addr#8,age#9L,name#10,sex#11] json +- Relation [addr#8,age#9L,name#10,sex#11] json22/04/01 22:28:41 INFO PlanChangeLogger:
=== Applying Rule org.apache.spark.sql.catalyst.analysis.ResolveTimeZone ===Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11] +- Project [addr#8, age#9L, name#10, sex#11]+- SubqueryAlias t_user +- SubqueryAlias t_user+- View (`t_user`, [addr#8,age#9L,name#10,sex#11]) +- View (`t_user`, [addr#8,age#9L,name#10,sex#11])+- Relation [addr#8,age#9L,name#10,sex#11] json +- Relation [addr#8,age#9L,name#10,sex#11] json22/04/01 22:28:41 INFO PlanChangeLogger:
=== Result of Batch Resolution ===
!'Project [unresolvedalias(cast(addr#8 as string), None), unresolvedalias(cast(age#9L as string), None), unresolvedalias(cast(name#10 as string), None), unresolvedalias(cast(sex#11 as string), None)] Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11] +- Project [addr#8, age#9L, name#10, sex#11]+- SubqueryAlias t_user +- SubqueryAlias t_user+- View (`t_user`, [addr#8,age#9L,name#10,sex#11]) +- View (`t_user`, [addr#8,age#9L,name#10,sex#11])+- Relation [addr#8,age#9L,name#10,sex#11] json +- Relation [addr#8,age#9L,name#10,sex#11] json22/04/01 22:28:41 INFO PlanChangeLogger: Batch Remove TempResolvedColumn has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch Apply Char Padding has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch Post-Hoc Resolution has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch Remove Unresolved Hints has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch Nondeterministic has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch UDF has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch UpdateNullability has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch Subquery has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch Cleanup has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger: Batch HandleAnalysisOnlyCommand has no effect.
22/04/01 22:28:41 INFO PlanChangeLogger:
=== Metrics of Executed Rules ===
Total number of runs: 126
Total time: 1.524472389 seconds
Total number of effective runs: 2
Total time of effective runs: 1.517592556 seconds
上面的逻辑计划触发的解析规则有
- ResolveAliases
- ResolveTimeZone
最终再次经历了analysis阶段的逻辑计划长这样：
Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [addr#8, age#9L, name#10, sex#11] +- SubqueryAlias t_user +- View (`t_user`, [addr#8,age#9L,name#10,sex#11]) +- Relation [addr#8,age#9L,name#10,sex#11] json
take
在 Dataset.select 之后就紧接着触发了Dataset.take。
def take(n: Int): Array[T] = head(n)
运行 take 需要将数据移动到应用程序的 Driver 进程中，如果使用非常大的 n，可能会导致 Driver 进程崩溃，并导致 OutOfMemoryError。
head
Dataset.take 方法是通过 Dataset.head 来实现的。
def head(n: Int): Array[T] = withAction("head", limit(n).queryExecution)(collectFromPlan)
同上面的 take，只有当生成的数组预计很小时，才应使用 head 方法，因为所有数据都加载到 Driver 的内存中。
limit
Dataset.take 方法底层是在 Dataset.limit 方法上面做了一个封装。
def limit(n: Int): Dataset[T] = withTypedPlan {Limit(Literal(n), logicalPlan)}
scala 中默认会调用 object Limit 的apply方法
def apply(limitExpr: Expression, child: LogicalPlan): UnaryNode = {GlobalLimit(limitExpr, LocalLimit(limitExpr, child))}
我们再来看看GlobalLimit 和 LocalLimit 是什么？
GlobalLimit & LocalLimit
/*** 一个全局的 (协同) limit. * 此运算符总共最多可以发出`limitExpr`个。*/
case class GlobalLimit(limitExpr: Expression, child: LogicalPlan) extends OrderPreservingUnaryNode {// 输出override def output: Seq[Attribute] = child.output// 最大行数override def maxRows: Option[Long] = {limitExpr match {case IntegerLiteral(limit) => Some(limit)case _ => None}}final override val nodePatterns: Seq[TreePattern] = Seq(LIMIT)override protected def withNewChildInternal(newChild: LogicalPlan): GlobalLimit =copy(child = newChild)
}/*** 一个分区局部的 (非协同) limit. * 每个物理分区上这个运算符最多可以发出`limitExpr`个。*/
case class LocalLimit(limitExpr: Expression, child: LogicalPlan) extends OrderPreservingUnaryNode {override def output: Seq[Attribute] = child.outputoverride def maxRowsPerPartition: Option[Long] = {limitExpr match {case IntegerLiteral(limit) => Some(limit)case _ => None}}final override val nodePatterns: Seq[TreePattern] = Seq(LIMIT)override protected def withNewChildInternal(newChild: LogicalPlan): LocalLimit =copy(child = newChild)
}
Dataset.limit 方法通过获取前 n 行返回新的数据集。
这个方法和
head之间的区别在于head是一个 action，返回一个数组（通过触发查询执行），而limit返回一个新的数据集。
变化
逻辑计划变化前：
Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]
+- Project [addr#8, age#9L, name#10, sex#11]+- SubqueryAlias t_user+- View (`t_user`, [addr#8,age#9L,name#10,sex#11])+- Relation [addr#8,age#9L,name#10,sex#11] json
逻辑计划变化后：
GlobalLimit 21
+- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11]+- SubqueryAlias t_user+- View (`t_user`, [addr#8,age#9L,name#10,sex#11])+- Relation [addr#8,age#9L,name#10,sex#11] json
上面的 21 表示的最大（分区）数据行数，这个数字来自Dataset.getRows的这行代码：
val data = newDf.select(castCols: _*).take(numRows + 1)
我们知道 Dataset.show 默认返回前 20 行数据，即numRows 默认就是 20，故 21 这个数字是符合我们的预期的。
而且，根据Limit.apply的这行代码可知：
GlobalLimit(limitExpr, LocalLimit(limitExpr, child))
GlobalLimit 为根节点，下面的子节点是 LocalLimit，再下面的子节点才是之前的逻辑计划，整个逻辑计划树的层次结构也是完全 OK 的。
withAction
Dataset.head 方法在调用了Dataset.limit之后，会调用Dataset.withAction，我们看看它是怎么实现的：
阅读源码的过程就是抽丝剥茧，我们现在做的就是打开外面的包装，找到最核心的内容。
/*** 包装数据集操作以跟踪查询执行和时间开销，然后向用户注册的回调函数报告。*/private def withAction[U](name: String, qe: QueryExecution)(action: SparkPlan => U) = {SQLExecution.withNewExecutionId(qe, Some(name)) {qe.executedPlan.resetMetrics()action(qe.executedPlan)}}
看到这个闭包结构实际上就已经很明确了，核心的内容就在里面。
Apache Spark 源码中类似的闭包结构随处可见，一般情况下我们不用过多关注外层的封装，这些都是用于处理一些额外操作比如收集统计信息等。
QueryExecution.executedPlan
/*** executedPlan 不应该用于初始化一切 SparkPlan，它应该只能用来执行*/lazy val executedPlan: SparkPlan = withCteMap {// 此处我们需要物化 optimizedPlan，在追踪 planning 阶段之前，要确定优化的时间不会计算到 planning 阶段里面assertOptimized()executePhase(QueryPlanningTracker.PLANNING) {// 克隆计划来避免在不同的阶段比如：analyzing/optimizing/planning 之间共享计划实例QueryExecution.prepareForExecution(preparations, sparkPlan.clone())}}
此处，我们顺便找到了 planning 阶段的入口，很明显，optimization 阶段执行完就立马执行 planning 阶段了
private def assertOptimized(): Unit = optimizedPlan
lazy val optimizedPlan: LogicalPlan = {// 此处，我们需要物化 commandExecuted，这样也不会计算到 optimization 阶段时间里面assertCommandExecuted()executePhase(QueryPlanningTracker.OPTIMIZATION) {// 克隆计划来避免在不同的阶段比如：analyzing/optimizing/planning 之间共享计划实例val plan =sparkSession.sessionState.optimizer.executeAndTrack(withCachedData.clone(), tracker)// 我们不希望优化的计划被重新分析为文本值，这样会导致常量折叠，在分析过程中会导致问题。而`clone`应该保持逻辑计划的`analyzed`状态，所以我们偏执地将计划设置为 `analyzed`plan.setAnalyzed()plan}}
QueryExecution.commandExecuted
在执行 optimization 阶段之前，还要完成一个关键的步骤：commandExecuted
lazy val commandExecuted: LogicalPlan = mode match {case CommandExecutionMode.NON_ROOT => analyzed.mapChildren(eagerlyExecuteCommands)case CommandExecutionMode.ALL => eagerlyExecuteCommands(analyzed)case CommandExecutionMode.SKIP => analyzed}
上面的 mode 来自枚举类CommandExecutionMode
这是啥？有什么意义呢？
CommandExecutionMode 表示命令执行的模式，总共有 3 种模式：ALL（默认），NON_ROOT，SKIP。
在 Spark 3.x 版本以前，我们使用 sql("INSERT…") 是不会立即触发表插入的，此时必须还得跟上 .collect() 来触发。
但是现在有了 CommandExecutionMode 就不用那么麻烦了，默认情况下 ALL 就可以支持整棵树以前序遍历的方式立即执行命令。
为了避免无止境地递归，在递归执行命令时我们应该使用NON_ROOT。
请注意，我们不能使用叶命令节点执行查询计划，因为许多命令返回
GenericInternalRow，不能直接放入查询计划中，否则查询引擎可能会将GenericInternalRow强制转换为UnsafeRow导致失败。
所谓的叶命令节点指的是那些没有其他依赖的命令节点，一般和数据源相关。
当运行 EXPLAIN 或其他命令中的命令时，我们应该使用 SKIP 来避免立即触发命令执行。
命令的立即执行是如何实现的呢？
private def eagerlyExecuteCommands(p: LogicalPlan) = p transformDown {case c: Command =>val qe = sparkSession.sessionState.executePlan(c, CommandExecutionMode.NON_ROOT)val result = SQLExecution.withNewExecutionId(qe, Some(commandExecutionName(c))) {qe.executedPlan.executeCollect()}CommandResult(qe.analyzed.output,qe.commandExecuted,qe.executedPlan,result)case other => other}
def executeCollect(): Array[InternalRow] = {// 将 unsaferow 打包到字节数组中，以实现更快的序列化。val byteArrayRdd = getByteArrayRdd()val results = ArrayBuffer[InternalRow]()// 底层还是 RDD.collect()byteArrayRdd.collect().foreach { countAndBytes =>decodeUnsafeRows(countAndBytes._2).foreach(results.+=)}results.toArray}
故，所谓的立即执行其实就是根据物理计划生成 RDD[InternalRow]，然后调用 “RDD.collect()” 而已，底层还是 collect 只不过不需要我们手动在代码中写明。
小结
上面的源码解析七绕八绕的，很多同学看起来可能就蒙了。
故我们先简单做个小结，从Dataset.show开始，到我们最后找到optimization阶段的入口，到底发生了什么？
整体的流程图如下：
从上面的思维导图可以看出，show() 方法总体分成了 2 个部分：
- 数据获取
- 字符串格式化
其中数据获取也分为 2 个步骤：
- select：由于需要输出到控制台，故除了 Binary 类型外，所有的字段类型都会转化成字符串，select 就是负责从逻辑计划中
挑选出想要的字段。这个过程会涉及到类型强转和字段别名，对于未解析的别名会触发analysis阶段应用解析规则来解析。 - take：take 想做的事情就是从最终返回的数据行中
拿出前 N 行数据，take 底层调用的是 head，而 head 底层是通过调用 limit 来实现的，只不过 head 返回的是数组，而 limit 返回的是逻辑计划。limit 返回的逻辑计划需要触发optimization阶段和planning阶段才能得到最终的数据行。
进入正题
接下来，就进入正题了，也就是我们的 optimization 阶段开始了。
executePhase(QueryPlanningTracker.OPTIMIZATION) {val plan =
sparkSession.sessionState.optimizer.executeAndTrack(withCachedData.clone(), tracker)plan.setAnalyzed()plan}
withCachedData
可以看到，上面传入的逻辑计划实际上是 withCachedData 的副本，那么withCachedData到底是什么呢？
lazy val withCachedData: LogicalPlan = sparkSession.withActive {// 用来检测逻辑计划是否已经经过`analysis`阶段assertAnalyzed()// 用来检测流数据查询的assertSupported()// 克隆计划来避免在不同的阶段比如：analyzing/optimizing/planning 之间共享计划实例
sparkSession.sharedState.cacheManager.useCachedData(commandExecuted.clone())}
故，optimization 阶段的第一步就是要去 CacheManager 里面走一遭~
那么，CacheManager到底是个啥？
请参考我的这篇博客——一篇文章搞懂 Spark 3.x 的 CacheManager
optimization 阶段首先要去CacheManager保存的缓存里面查一查有没有已缓存的查询计划，有的话复用就可以了，没必要再跑一遍。
executeAndTrack
接下来的源码大家可能比较熟悉了，实际上我们在上一讲中已经讲到了这部分源码，只不过上一讲中的主体是 RuleExecutor 的一个子类 Analyzer，而我们这一讲的主体是RuleExecutor 的另一个子类
SparkOptimizer，它们共用了父类RuleExecutor中的方法executeAndTrack 和 execute。
def executeAndTrack(plan: TreeType, tracker: QueryPlanningTracker): TreeType = {QueryPlanningTracker.withTracker(tracker) {execute(plan)}}
忘记的同学可以回头看看前面的源码解析，这里简单介绍一下executeAndTrack 和 execute 是怎么实现的：
executeAndTrack 底层调用的就是 execute，只不过封装了一些追踪的逻辑。
execute 就是针对 batches 即规则批中的每一个规则都调用它的 apply 方法针对逻辑计划做出一定的调整，直到达到固定的次数或者逻辑计划不再发生改变时停止。
execute 的核心其实就在于一个个的规则！
我们来看看 optimization 阶段的规则有哪些？和 analysis阶段有什么不同？
batches
SparkOptimizer 的 batches 定义在 Optimizer.scala 文件里面。
final override def batches: Seq[Batch] = {val excludedRulesConf =SQLConf.get.optimizerExcludedRules.toSeq.flatMap(Utils.stringToSeq)val excludedRules = excludedRulesConf.filter { ruleName =>val nonExcludable = nonExcludableRules.contains(ruleName)if (nonExcludable) {logWarning(s"Optimization rule '${ruleName}' was not excluded from the optimizer " +s"because this rule is a non-excludable rule.")}!nonExcludable}if (excludedRules.isEmpty) {defaultBatches} else {defaultBatches.flatMap { batch =>val filteredRules = batch.rules.filter { rule =>val exclude = excludedRules.contains(rule.ruleName)if (exclude) {logInfo(s"Optimization rule '${rule.ruleName}' is excluded from the optimizer.")}!exclude}if (batch.rules == filteredRules) {Some(batch)} else if (filteredRules.nonEmpty) {Some(Batch(batch.name, batch.strategy, filteredRules: _*))} else {logInfo(s"Optimization batch '${batch.name}' is excluded from the optimizer " +s"as all enclosed rules have been excluded.")None}}}}
}
可以看到，Optimizer 将规则批分成了 3 种类型：
- defaultBatches
- excludedRules——通过 Spark 配置参数可以排除一些规则阻止其进行优化
- nonExcludableRules——即使 Spark 配置参数指定了也无法排除的优化规则
重点在于默认的规则批——defaultBatches，为了实现上的可扩展性，我们将标准的优化规则放到了抽象类 Optimizer 中，特殊的规则放到了SparkOptimizer中。
标准的优化规则
算子优化前
| 规则 | 所属规则批 | 执行策略 | 解释说明 | 补充说明 |
|---|---|---|---|---|
| EliminateDistinct | Eliminate Distinct | Once |
删移除关于MAX和MIN的无效DISTINCT。在 RewriteDistinctAggregates 之前，应该先应用此规则。
|
|
| EliminateResolvedHint | Finish Analysis | Once |
替换计划中的ResolvedHint算子。将HintInfo移动到关联的Join算子，否则，如果没有匹配的Join算子，就将其删除。
|
HintInfo 是要应用于特定节点的提示属性
|
| EliminateSubqueryAliases | Finish Analysis | Once |
消除子查询别名，对应逻辑算子树中的SubqueryAlias节点。一般来讲，Subqueries 仅用于提供查询的视角范围(Scope)信息，一旦 analysis 阶段结束， 该节点就可以被移除，该优化规则直接将SubqueryAlias替换为其子节点。
|
|
| EliminateView | Finish Analysis | Once |
此规则将从计划中删除View算子。在analysis阶段结束之前，这个算子会一直受到尊重，因为我们希望看到AnalyzedLogicalPlan的哪一部分是从视图生成的。
|
|
| InlineCTE | Finish Analysis | Once | 如果满足以下任一条件，则将CTE定义插入相应的引用中：1. CTE定义不包含任何非确定性表达式。如果此CTE定义引用了另一个具有非确定性表达式的CTE定义，则仍然可以内联当前CTE定义。2.在整个主查询和所有子查询中，CTE定义只被引用一次。 此外，由于相关子查询的复杂性，无论上述条件如何，相关子查询中的所有CTE引用都是内联的。 | |
| ReplaceExpressions | Finish Analysis | Once |
查找所有无法执行的表达式，并用可计算的语义等价表达式替换/重写它们。目前，我们替换了两种表达式：1.RuntimeReplaceable表达式。 2.无法执行的聚合表达式，如Every/Some/Any/CountIf 这主要用于提供与其他数据库的兼容性。很少有这样的例子：我们使用它来支持nvl，将其替换为coalesce。我们分别用Min和Max替换each和Any。
|
|
| RewriteNonCorrelatedExists | Finish Analysis | Once |
为了使用ScalarSubquery需要重写非关联的exists子查询，比如：WHERE EXISTS (SELECT A FROM TABLE B WHERE COL1 > 10) 会被重写成 WHERE (SELECT 1 FROM (SELECT A FROM TABLE B WHERE COL1 > 10) LIMIT 1) IS NOT NULL
|
ScalarSubquery是只返回一行和一列的子查询。这将在planning阶段转换为物理标量（scalar）子查询。
|
| PullOutGroupingExpressions | Finish Analysis | Once |
此规则确保Aggregate节点在optimization阶段不包含复杂的分组表达式。 复杂的分组表达式被拉到Aggregate下的Project节点，并在分组表达式和不带聚合函数的聚合表达式中引用。这些引用确保优化规则不会将聚合表达式更改为不再引用任何分组表达式的无效表达式，并简化节点上的表达式转换（只需转换表达式一次）。例如，在下面的查询中，Spark不应该将聚合表达式Not(IsNull(c))优化成IsNotNull(c)，因为IsNull(c)是一个分组表达式：SELECT not(c IS NULL) FROM t GROUP BY c IS NULL
|
|
| ComputeCurrentTime | Finish Analysis | Once | 计算当前日期和时间，以确保在单个查询中返回相同的结果。 | |
| ReplaceCurrentLike | Finish Analysis | Once |
用当前数据库名称替换CurrentDatabase的表达式。用当前catalog名称替换CurrentCatalog的表达式。
|
|
| SpecialDatetimeValues | Finish Analysis | Once | 如果输入字符串是可折叠的，则用其日期/时间戳值强制转换成特殊日期时间字符串。 | |
| RewriteAsOfJoin | Finish Analysis | Once |
使用 Join 和 Aggregate 运算符的组合替换逻辑AsOfJoin运算符。
|
|
| RemoveNoopOperators | Union | Once | 从查询计划中删除不进行任何修改的 no-op 运算符。 | |
| CombineUnions | Union | Once |
将所有相邻的Union运算符合并成一个
|
|
| RemoveNoopUnion | Union | Once |
简化 Union 的子节点，或者从查询计划中删除不修改查询的 no-op Union。
|
|
| OptimizeLimitZero | OptimizeLimitZero | Once |
将GlobalLimit 0和LocalLimit 0节点（子树）替换为空的Local Relation，因为它们不返回任何数据行。
|
|
| ConvertToLocalRelation | LocalRelation early | fixedPoint |
将LocalRelation上的本地操作（即不需要数据交换的操作）转换为另一个LocalRelation。
|
|
| PropagateEmptyRelation | LocalRelation early | fixedPoint | 简化了空或非空关系的查询计划。 |
当删除一个Union空关系子级时，PropagateEmptyRelation可以将属性（attribute）的可空性从可空更改为非空
|
| UpdateAttributeNullability | LocalRelation early | fixedPoint |
通过使用其子输出属性（Attributes）的相应属性的可空性，更新已解析LogicalPlan中属性的可空性。之所以需要此步骤，是因为用户可以在Dataset API中使用已解析的AttributeReference，而outer join可以更改AttributeReference的可空性。如果没有这个规则，可以为NULL的列的NULL字段实际上可以设置为non-NULL，这会导致非法优化（例如NULL传播）和错误的答案。
|
|
| OptimizeOneRowRelationSubquery | Pullup Correlated Expressions | Once |
此规则优化将OneRowRelation作为叶节点的子查询。
|
|
| PullupCorrelatedPredicates | Pullup Correlated Expressions | Once |
从给定的子查询中取出所有（外部）相关谓词。此方法从子查询Filter中删除相关谓词，并将这些谓词的引用添加到所有中间Project和Aggregate子句（如果缺少的话），以便能够在顶层评估谓词。
|
|
| OptimizeSubqueries | Subquery | FixedPoint(1) | 优化表达式中的所有子查询。 |
子查询批处理递归地应用优化器规则。因此，在其上强制幂等性这毫无意义，我们将这个批次从Once改成了FixedPoint(1)。
|
| RewriteExceptAll | Replace Operators | fixedPoint |
混合使用Union、Aggregate、Generate 运算符来替代逻辑的Except运算符。
|
|
| RewriteIntersectAll | Replace Operators | fixedPoint |
混合使用Union、Aggregate、Generate 运算符来替代逻辑的Intersect运算符。
|
|
| ReplaceIntersectWithSemiJoin | Replace Operators | fixedPoint |
使用 left-semi Join 运算符替代逻辑Intersect运算符。
|
|
| ReplaceExceptWithFilter | Replace Operators | fixedPoint |
如果逻辑Except运算符中的一或两个数据集都纯粹地使用Filter转换过，这个规则会使用反转Except运算符右侧条件之后的Filter运算符替代。
|
|
| ReplaceExceptWithAntiJoin | Replace Operators | fixedPoint |
使用 left-anti Join运算符替代逻辑Except运算符。
|
|
| ReplaceDistinctWithAggregate | Replace Operators | fixedPoint |
使用Aggregate运算符替代逻辑Distinct运算符。
|
|
| ReplaceDeduplicateWithAggregate | Replace Operators | fixedPoint |
使用Aggregate运算符替代逻辑Deduplicate运算符。
|
|
| RemoveLiteralFromGroupExpressions | Aggregate | fixedPoint |
移除Aggregate运算符中分组表达式的文本值，因为它们除了使得分组键变得更大以外，对结果没有任何影响。
|
|
| RemoveRepetitionFromGroupExpressions | Aggregate | fixedPoint |
移除Aggregate运算符中分组表达式的重复内容，因为它们除了使得分组键变得更大以外，对结果没有任何影响。
|
注意点
Finish Analysis
从技术上讲，Finish Analysis中的一些规则不是优化器规则，而是更应该属于 Analyzer，因为它们是正确性所必需的（例如ComputeCurrentTime）。
但是，因为我们也使用Analyzer来规范化查询（用于视图定义），但是我们不会在Analyzer中消除子查询或计算当前时间。
Union
优化规则是从 Union 开始的
- 在启动主要优化器规则之前，首先调用
CombineUnions，因为它可以减少迭代次数，而其他规则可以增加/移除两个相邻 Union 运算符之间的额外运算符。 - 在规则批 Batch(“Operator Optimizations”)中会再次调用
CombineUnions，因为其他规则可能会使两个独立的Union运算符相邻。
LocalRelation early
提前运行一次。这可能会简化计划并降低优化器的成本。
例如，像Filter(LocalRelation)这样的查询当有一个filter被触发时会经历所有繁重的优化规则（比如InferFiltersFromConstraints）。
如果我们更早地运行这个批处理，查询就会变得非常简单，LocalRelation并不会触发许多规则。
Replace Operators
RewriteExceptAll
SELECT c1 FROM ut1 EXCEPT ALL SELECT c1 FROM ut2
会被重写成：
SELECT c1
FROM (SELECT replicate_rows(sum_val, c1)FROM (SELECT c1, sum_valFROM (SELECT c1, sum(vcol) AS sum_valFROM (SELECT 1L as vcol, c1 FROM ut1UNION ALLSELECT -1L as vcol, c1 FROM ut2) AS union_allGROUP BY union_all.c1)WHERE sum_val > 0)
)
RewriteIntersectAll
SELECT c1 FROM ut1 INTERSECT ALL SELECT c1 FROM ut2
会被重写成：
SELECT c1
FROM (SELECT replicate_row(min_count, c1)FROM (SELECT c1, If (vcol1_cnt > vcol2_cnt, vcol2_cnt, vcol1_cnt) AS min_countFROM (SELECT c1, count(vcol1) as vcol1_cnt, count(vcol2) as vcol2_cntFROM (SELECT true as vcol1, null as , c1 FROM ut1UNION ALLSELECT null as vcol1, true as vcol2, c1 FROM ut2) AS union_allGROUP BY c1HAVING vcol1_cnt >= 1 AND vcol2_cnt >= 1)))
ReplaceIntersectWithSemiJoin
SELECT a1, a2 FROM Tab1 INTERSECT SELECT b1, b2 FROM Tab2
会被重写成：
SELECT DISTINCT a1, a2 FROM Tab1 LEFT SEMI JOIN Tab2 ON a1<=>b1 AND a2<=>b2
ReplaceExceptWithFilter
SELECT a1, a2 FROM Tab1 WHERE a2 = 12 EXCEPT SELECT a1, a2 FROM Tab1 WHERE a1 = 5
会被重写成：
SELECT DISTINCT a1, a2 FROM Tab1 WHERE a2 = 12 AND (a1 is null OR a1 <> 5)
注意，在反转右节点的过滤条件之前，我们应该：
- 所有的
Filter运算符要合并到一起 - 更新属性（attribute）引用到左节点
- 添加一个
Coalesce(condition, False)（要考虑到条件中的空值情况）
ReplaceExceptWithAntiJoin
SELECT a1, a2 FROM Tab1 EXCEPT SELECT b1, b2 FROM Tab2
会被重写成：
SELECT DISTINCT a1, a2 FROM Tab1 LEFT ANTI JOIN Tab2 ON a1<=>b1 AND a2<=>b2
ReplaceDistinctWithAggregate
SELECT DISTINCT f1, f2 FROM t
会被重写成：
SELECT f1, f2 FROM t GROUP BY f1, f2
算子优化
过滤推断前的算子优化
算子下推
| 规则 | 解释说明 |
|---|---|
| PushProjectionThroughUnion |
将Project操作符推送到Union操作符的两侧。可安全下推的操作如下所示。Union：现在，Union就意味着Union ALL，它不消除重复行。因此，通过它下推Filter和Project是安全的。下推Filter是由另一个规则PushDownPredicates处理的。一旦我们添加了UNION DISTINCT，我们就无法下推Project了。
|
| ReorderJoin |
重新排列Join，并将所有条件推入Join，以便底部的条件至少有一个条件。 如果所有Join都已具有至少一个条件，则Join的顺序不会更改。 如果启用了星型模式检测，请基于启发式重新排序星型Join计划。
|
| EliminateOuterJoin |
1.消除outer join，前提是谓词可以限制结果集，以便消除所有空行：如果两侧都有这个谓词，full outer -> inner；如果右侧有这个谓词，left outer -> inner；如果左侧有这个谓词，right outer -> inner；当且仅当左侧有这个谓词，full outer -> left outer；当且仅当右侧有这个谓词，full outer -> right outer 2.如果outer join仅在流侧具有distinct，则移除outer join： SELECT DISTINCT f1 FROM t1 LEFT JOIN t2 ON t1.id = t2.id ==> SELECT DISTINCT f1 FROM t1。当前规则应该在谓词下推之前执行。
|
| PushDownPredicates |
常规的运算符和Join谓词下推的统一版本。此规则提高了级联join（例如：Filter-Join-Join-Join）的谓词下推性能。大多数谓词可以在一次传递中下推。
|
| PushDownLeftSemiAntiJoin |
这个规则是PushPredicateThroughNonJoin的一个变体，它可以下推以下运算符的Left semi join和Left Anti join：1.Project2.Window3.Union4.Aggregate5.其他允许的一元运算符。
|
| PushLeftSemiLeftAntiThroughJoin |
此规则是PushPredicateThrowJoin的一个变体，它可以下推join运算符下面的Left semi join和Left Anti join：允许的Join类型有：1.Inner；2.Cross；3.LeftOuter；4.RightOuter
|
| LimitPushDown |
下推UNION ALL和JOIN下的LocalLimit。
|
| LimitPushDownThroughWindow |
下推Window下方的LocalLimit。此规则优化了以下情况：SELECT *, ROW_NUMBER() OVER(ORDER BY a) AS rn FROM Tab1 LIMIT 5 ==> SELECT *, ROW_NUMBER() OVER(ORDER BY a) AS rn FROM (SELECT * FROM Tab1 ORDER BY a LIMIT 5) t
|
| ColumnPruning |
试图消除查询计划中不需要的列读取。 由于在Filter之前添加Project会和PushPredicatesThroughProject冲突，此规则将以以下模式删除Project p2： p1 @ Project(_, Filter(_, p2 @ Project(_, child))) if p2.outputSet.subsetOf(p2.inputSet) p2通常是按照这个规则插入的，没有用，p1无论如何都可以删减列。
|
| GenerateOptimization |
如果Generate位于不引用任何生成属性的Project下，则从中删除不必要的字段。例如，在一个分解的数组上类似count的聚合。
|
算子合并
| 规则 | 解释说明 |
|---|---|
| CollapseRepartition |
合并相邻的RepartitionOperation和RebalancePartitions运算符
|
| CollapseProject |
两个Project运算符合并为一个别名替换，在以下情况下，将表达式合并为一个表达式。1.两个Project运算符相邻时。2.当两个Project运算符之间有LocalLimit/Sample/Repartition运算符，且上层的Project由相同数量的列组成，且列数相等或具有别名时。同时也考虑到GlobalLimit(LocalLimit)模式。
|
| OptimizeWindowFunctions |
将first(col)替换成nth_value(col, 1)来获得更好的性能
|
| CollapseWindow |
折叠相邻的Window表达式。如果分区规格和顺序规格相同，并且窗口表达式是独立的，且属于相同的窗口函数类型，则折叠到父节点中。
|
| CombineFilters |
将两个相邻的Filter运算符合并为一个，将非冗余条件合并为一个连接谓词。
|
| EliminateLimits |
该规则由normal和AQE优化器应用，并通过以下方式优化Limit运算符：1.如果是child max row<=Limit，则消除Limit/GlobalLimit运算符。2.将两个相邻的Limit运算符合并为一个，将多个表达式合并成一个。
|
| CombineUnions |
将所有相邻的Union运算符合并为一个。
|
常量折叠和强度消减
在编译器构建中，强度消减（strength reduction）是一种编译器优化，其中昂贵的操作被等效但成本较低的操作所取代。强度降低的经典例子将循环中的“强”乘法转换为“弱”加法——这是数组寻址中经常出现的情况。强度消减的例子包括：用加法替换循环中的乘法、用乘法替换循环中的指数。
| 规则 | 解释说明 |
|---|---|
| OptimizeRepartition |
如果所有的分区表达式都被折叠了并且用户未指定的情况下，将RepartitionByExpression的分区数设置成 1
|
| TransposeWindow | 转置相邻的窗口表达式。如果父窗口表达式的分区规范与子窗口表达式的分区规范兼容，就转置它们。 |
| NullPropagation |
替换可以用等效Literal静态计算的Expression Expressions。对于从表达式树的底部到顶部的空值传播，这个规则会更加具体。
|
| NullDownPropagation |
如果输入不允许为空，则展开IsNull/IsNotNull的输入，例如：IsNull(Not(null)) == IsNull(null)
|
| ConstantPropagation |
用连接表达式中的相应值替换可以静态计算的属性，例如：SELECT * FROM table WHERE i = 5 AND j = i + 3 ==> SELECT * FROM table WHERE i = 5 AND j = 8 使用的方法：通过查看所有相等的谓词来填充属性 => 常量值的映射；使用这个映射，将属性的出现的地方替换为AND节点中相应的常量值。
|
| FoldablePropagation |
如果可能，用原始可折叠表达式的别名替换属性。其他优化将利用传播的可折叠表达式。例如，此规则可以将SELECT 1.0 x, 'abc' y, Now() z ORDER BY x, y, 3 优化成SELECT 1.0 x, 'abc' y, Now() z ORDER BY 1.0, 'abc', Now() 其他规则可以进一步优化它，并且删除ORDER BY运算符。
|
| OptimizeIn |
优化IN谓词：1.当列表为空且值不可为null时，将谓词转换为false。2.删除文本值重复。3.将In (value, seq[Literal])替换为更快的优化版本InSet (value, HashSet[Literal])。
|
| ConstantFolding | 替换可以用等效文本值静态计算的表达式。 |
| EliminateAggregateFilter |
删除聚合表达式的无用FILTER子句。在RewriteDistinctAggregates之前，应该先应用这个规则。
|
| ReorderAssociativeOperator | 重新排列相关的整数类型运算符，并将所有常量折叠为一个。 |
| LikeSimplification |
简化了不需要完整正则表达式来计算条件的LIKE表达式。例如，当表达式只是检查字符串是否以给定模式开头时。
|
| BooleanSimplification |
简化了布尔表达式：1.简化了答案可以在不计算双方的情况下确定的表达式。2.消除/提取共同的因子。3.合并相同的表达式4。删除Not运算符。
|
| SimplifyConditionals | 简化了条件表达式（if / case） |
| PushFoldableIntoBranches |
将可折叠表达式下推到if / case分支
|
| RemoveDispensableExpressions | 移除不必要的节点 |
| SimplifyBinaryComparison |
使用语义相等的表达式简化二进制比较：1.用true文本值替代<==>；2.如果操作数都是非空的，用true文本值替代 =， <=， 和 >=；3.如果操作数都是非空的，用false文本值替代>和<；4.如果有一边操作数是布尔文本值，就展开=、<=>
|
| ReplaceNullWithFalseInPredicate |
一个用FalseLiteral替换Literal(null, BooleanType) 的规则，如果可能的话，在WHERE/HAVING/ON(JOIN)子句的搜索条件中，该子句包含一个隐式布尔运算符(search condition) = TRUE。当计算整个搜索条件时，只有当Literal(null, BooleanType)在语义上等同于FalseLiteral时，替换才有效。 请注意，在大多数情况下，当搜索条件包含NOT和可空的表达式时，FALSE和NULL是不可交换的。因此，该规则非常保守，适用于非常有限的情况。 例如，Filter(Literal(null, BooleanType))等同于Filter(FalseLiteral)。 另一个包含分支的示例是Filter(If(cond, FalseLiteral, Literal(null, _)))；这可以优化为Filter(If(cond, FalseLiteral, FalseLiteral))，最终Filter(FalseLiteral)。 此外，该规则还转换所有If表达式中的谓词，以及所有CaseWhen表达式中的分支条件，即使它们不是搜索条件的一部分。 例如，Project(If(And(cond, Literal(null)), Literal(1), Literal(2)))可以简化为Project(Literal(2))。
|
| SimplifyConditionalsInPredicate |
一个规则，在WHERE/HAVING/ON(JOIN)子句的搜索条件中，如果可能，将条件表达式转换为谓词表达式，其中包含一个隐式布尔运算符(search condition) = TRUE。在这个转换之后，我们可以潜在地下推过滤到数据源。这个规则是安全可空的。支持的用例有：1.IF(cond, trueVal, false) => AND(cond, trueVal)；2.IF(cond, trueVal, true) => OR(NOT(cond), trueVal)；3.IF(cond, false, falseVal) => AND(NOT(cond), falseVal)；4.IF(cond, true, falseVal) => OR(cond, falseVal)；5.CASE WHEN cond THEN trueVal ELSE false END => AND(cond, trueVal)；6.CASE WHEN cond THEN trueVal END => AND(cond, trueVal)；7.CASE WHEN cond THEN trueVal ELSE null END => AND(cond, trueVal)；8.CASE WHEN cond THEN trueVal ELSE true END => OR(NOT(cond), trueVal)；9.CASE WHEN cond THEN false ELSE elseVal END => AND(NOT(cond), elseVal)；10.CASE WHEN cond THEN true ELSE elseVal END => OR(cond, elseVal)
|
| PruneFilters |
删除可以进行简单计算的Filter。这可以通过以下方式实现：1.在其计算结果始终为true的情况下，省略Filter。2.当筛选器的计算结果总是为false时，替换成一个伪空关系。3.消除子节点输出给定约束始终为true的条件。
|
| SimplifyCasts | 删除不必要的强制转换，因为输入已经是正确的类型。 |
| SimplifyCaseConversionExpressions | 删除不必要的内部大小写转换表达式，因为内部转换被外部转换覆盖。 |
| RewriteCorrelatedScalarSubquery |
此规则将相关的ScalarSubquery表达式重写为LEFT OUTER JOIN。
|
| RewriteLateralSubquery |
将LateralSubquery表达式重写成join
|
| EliminateSerialization | 消除不必要地在对象和数据项的序列化（InternalRow）表示之间切换的情况。例如，背对背映射操作。 |
| RemoveRedundantAliases | 从查询计划中删除冗余别名。冗余别名是不会更改列的名称或元数据，也不会消除重复数据的别名。 |
| RemoveRedundantAggregates | 从查询计划中删除冗余聚合。冗余聚合是一种聚合，其唯一目标是保持不同的值，而其父聚合将忽略重复的值。 |
| UnwrapCastInBinaryComparison |
在二进制比较或In/InSet操作中使用如下模式进行展开强制转换：1. BinaryComparison(Cast(fromExp, toType), Literal(value, toType))；2.BinaryComparison(Literal(value, toType), Cast(fromExp, toType))；3.In(Cast(fromExp, toType), Seq(Literal(v1, toType), Literal(v2, toType), ...)；4.InSet(Cast(fromExp, toType), Set(v1, v2, ...)) 该规则通过使用更简单的构造替换强制转换，或者将强制转换从表达式端移动到文本值端，从而使用上述模式优化表达式，这使它们能够在以后进行优化，并向下推送到数据源。
|
| RemoveNoopOperators |
从查询计划中删除不进行任何修改的no-op运算符。
|
| OptimizeUpdateFields |
优化UpdateFields表达式链。
|
| SimplifyExtractValueOps |
简化冗余的CreateNamedStruct/CreateArray/CreateMap表达式
|
| OptimizeCsvJsonExprs |
简化冗余的csv/json相关表达式。 优化内容包括：1.JsonToStructs(StructsToJson(child)) => child.。2.删除GetStructField/GetArrayStructFields + JsonToStructs中不必要的列。3.CreateNamedStruct(JsonToStructs(json).col1, JsonToStructs(json).col2, ...) => If(IsNull(json), nullStruct, KnownNotNull(JsonToStructs(prunedSchema, ..., json)))如果JsonToStructs(json)在CreateNamedStruct的所有字段中共享。prunedSchema包含原始CreateNamedStruct中所有访问的字段。4.从GetStructField+CsvToStructs中删除不必要的列。
|
| CombineConcats |
合并嵌套的Concat表达式
|
Spark 还支持自定义优化规则，详情可以查看我的这篇博客——Spark SQL 如何自定义扩展？
过滤推断
| 规则 | 解释说明 |
|---|---|
| InferFiltersFromGenerate |
从Generate推断Filter，这样就可以在join之前和数据源中更早地通过这个Generate删除数据行。
|
| InferFiltersFromConstraints |
基于运算符的现有约束生成附加过滤器的列表，但删除那些已经属于运算符条件一部分或属于运算符子节点约束一部分的过滤器。这些筛选器当前插入到Filter运算符的和Join运算符任一侧的现有条件中。 注意：虽然这种优化适用于许多类型的join，但它主要有利于Inner Join和LeftSemi Join。
|
过滤推断后的算子优化
同上面过滤推断前的算子优化，这也说明了过滤会对算子的优化造成影响，故需要重新执行一遍算子优化规则。
下推 join 的额外谓词
| 规则 | 解释说明 |
|---|---|
| PushExtraPredicateThroughJoin |
尝试将JOIN条件下推到左和右子节点中。为了避免扩展JOIN条件，即使发生谓词下推，JOIN条件也将保持原始形式。
|
| PushDownPredicates |
常规的运算符和Join谓词下推的统一版本。此规则提高了级联join（例如：Filter-Join-Join-Join）的谓词下推性能。大多数谓词可以在一次传递中下推。
|
下推
JOIN条件之后再次执行谓词下推的规则。
依赖统计数据的优化规则
| 规则 | 所属规则批 | 执行策略 | 解释说明 | 补充说明 |
|---|---|---|---|---|
| preCBORules | Pre CBO Rules | Once | 这个规则批在算子优化后、在任何依赖统计数据的规则批之前重写了逻辑计划。 | |
| earlyScanPushDownRules | Early Filter and Projection Push-Down | Once |
这个规则批将Filter和Project下推到Scan节点。在这个规则批之前，逻辑计划可能包含不报告统计数据的节点。任何使用统计数据的规则都必须在这个规则批之后运行。
|
|
| UpdateCTERelationStats | Update CTE Relation Stats | Once |
更新CTE引用的统计信息。
|
|
| CostBasedJoinReorder | Join Reorder | FixedPoint(1) |
基于成本的JOIN重新排序。未来我们可能会有几种JOIN重排序算法。这个类是这些算法的入口，并选择要使用的算法。
|
由于AQP中的连接成本可能在多次运行之间发生变化，因此我们没有理由强制这个规则批上面的幂等性。因此，我们将其定为FixedPoint(1)，而不是Once。
|
| EliminateSorts | Eliminate Sorts | Once |
如果排序操作不影响最终的输出顺序，则删除它们。请注意，最终输出顺序的更改可能会影响文件大小。这个规则处理下面的情况：1.如果子节点的最大行数小于或等于1；2.如果排序顺序为空或排序顺序没有任何引用；3.如果排序运算符是本地排序且子节点已排序；4.如果有另一个排序运算符被 0...n 个 Project、Filter、Repartition或RepartitionByExpression、RebalancePartitions（使用确定性表达式）运算符分隔；5.如果排序运算符位于JOIN内，被0...n 个 Project、Filter、Repartition或RepartitionByExpression、RebalancePartitions（使用确定性表达式）运算符分隔，且只有JOIN条件是确定性的；6.如果排序运算符位于GroupBy中，被0...n 个 Project、Filter、Repartition或RepartitionByExpression、RebalancePartitions（使用确定性表达式）运算符分隔，且只有聚合函数是顺序无关的。
|
|
| DecimalAggregates | Decimal Optimizations | fixedPoint |
通过在未标度的长整型值上执行固定精度小数来加速聚合。 它使用和DecimalPrecision相同的规则来提高输出的精度和规模。
|
Hive_Decimal_Precision_Scale_Support |
| RewriteDistinctAggregates | Distinct Aggregate Rewrite | Once | 此规则将具有不同聚合的聚合查询重写为扩展的双精度查询聚合，其中正则聚合表达式和每个不同的子句被聚合单独一组。然后将结果合并到第二个聚合中。 |
此批处理必须在Decimal Optimizations之后运行，因为这样可能会更改aggregate distinct列
|
| EliminateMapObjects | Object Expressions Optimization | fixedPoint |
满足以下条件时删除MapObjects：1.Mapobject(... lambdavariable(..., false) ...)，这意味着输入和输出的类型都是非空原始类型；2.没有自定义集合类指定数据项的表示形式。
|
MapObjects将给定表达式应用于集合项的每个元素，并将结果作为ArrayType或ObjectType返回。这类似于典型的映射操作，但lambda函数是使用catalyst表达式表示的。
|
| CombineTypedFilters | Object Expressions Optimization | fixedPoint |
将两个相邻的TypedFilter（它们在条件下对同一类型对象进行操作）合并为一个，将筛选函数合并为一个连接函数。
|
TypedFilter将func应用于子元素的每个元素并按最终产生的布尔值过滤它们。这在逻辑上等于一个普通的Filter运算符，其条件表达式将输入行解码为对象，并将给定函数应用于解码的对象。然而，我们需要对TypedFilter进行封装，以使概念更清晰，并使编写优化器规则更容易。
|
| ObjectSerializerPruning | Object Expressions Optimization | fixedPoint | 从查询计划中删除不必要的对象序列化程序。此规则将删除序列化程序中的单个序列化程序和嵌套字段。 | |
| ReassignLambdaVariableID | Object Expressions Optimization | fixedPoint |
将每个查询的唯一ID重新分配给LambdaVariables，其原始ID是全局唯一的。这有助于Spark更频繁地访问codegen缓存并提高性能。
|
LambdaVariables是MapObjects中使用的循环变量的占位符。不应该手动构造，而是将其传递到提供的lambda函数中。
|
| ConvertToLocalRelation | Object Expressions Optimization | fixedPoint |
将LocalRelation上的本地操作（即不需要数据交换的操作）转换为另一个LocalRelation。
|
LocalRelation是用于扫描本地集合中的数据的逻辑计划节点。ConvertToLocalRelation 在上面的LocalRelation early规则批中也出现过。
|
| PropagateEmptyRelation | LocalRelation | fixedPoint | 简化了空或非空关系的查询计划。 |
当删除一个Union空关系子级时，PropagateEmptyRelation可以将属性（attribute）的可空性从可空更改为非空
|
| UpdateAttributeNullability | LocalRelation | fixedPoint |
通过使用其子输出属性（Attributes）的相应属性的可空性，更新已解析LogicalPlan中属性的可空性。之所以需要此步骤，是因为用户可以在Dataset API中使用已解析的AttributeReference，而outer join可以更改AttributeReference的可空性。如果没有这个规则，可以为NULL的列的NULL字段实际上可以设置为non-NULL，这会导致非法优化（例如NULL传播）和错误的答案。
|
UpdateAttributeNullability 在上面的LocalRelation early规则批中也出现过。
|
| CheckCartesianProducts | Check Cartesian Products | Once |
检查优化计划树中任何类型的join之间是否存在笛卡尔积。如果在未显示指定cross join的情况下找到笛卡尔积，则引发错误。如果CROSS_JOINS_ENABLED标志为true，则此规则将被有效禁用。 此规则必须在ReordJoin规则之后运行，因为在检查每个join是否为笛卡尔积之前，必须收集每个join的join条件。如果有SELECT * from R, S where R.r = S.s，则R和S之间的连接不是笛卡尔积，因此应该允许。谓词R.r=S.s在ReorderJoin规则之前不会被识别为join条件。 此规则必须在批处理LocalRelation之后运行，因为具有空关系的join不应是笛卡尔积。
|
从这往下的规则批都应该在规则批Join Reorder和LocalRelation之后执行。
|
| RewritePredicateSubquery | RewriteSubquery | Once |
这个规则将谓词子查询重写为left semi/anti join。支持以下谓词：1.EXISTS/NOT EXISTS将被重写为semi/anti join，Filter中未解析的条件将被提取为join条件。2.IN/NOT IN将被重写为semi/anti join，Filter中未解析的条件将作为join条件被拉出，value=selected列也将用作join条件。
|
|
| ColumnPruning | RewriteSubquery | Once |
试图消除查询计划中不需要的列读取。 由于在Filter之前添加Project会和PushPredicatesThroughProject冲突，此规则将以以下模式删除Project p2： p1 @ Project(_, Filter(_, p2 @ Project(_, child))) if p2.outputSet.subsetOf(p2.inputSet) p2通常是按照这个规则插入的，没有用，p1无论如何都可以删减列。
|
ColumnPruning规则在上面的算子优化中出现过。
|
| CollapseProject | RewriteSubquery | Once |
两个Project运算符合并为一个别名替换，在以下情况下，将表达式合并为一个表达式。1.两个Project运算符相邻时。2.当两个Project运算符之间有LocalLimit/Sample/Repartition运算符，且上层的Project由相同数量的列组成，且列数相等或具有别名时。同时也考虑到GlobalLimit(LocalLimit)模式。
|
CollapseProject规则在上面的算子优化中出现过。
|
| RemoveRedundantAliases | RewriteSubquery | Once | 从查询计划中删除冗余别名。冗余别名是不会更改列的名称或元数据，也不会消除重复数据的别名。 |
RemoveRedundantAliases规则在上面的算子优化中出现过。
|
| RemoveNoopOperators | RewriteSubquery | Once |
从查询计划中删除不进行任何修改的no-op运算符。
|
RemoveNoopOperators规则在上面的算子优化中出现过。
|
| NormalizeFloatingNumbers | NormalizeFloatingNumbers | Once |
这个规则规范化了窗口分区键、join key和聚合分组键中的NaN和-0.0。
|
这个规则必须在RewriteSubquery之后运行，后者会创建join
|
| ReplaceUpdateFieldsExpression | ReplaceUpdateFieldsExpression | Once |
用可计算表达式替换UpdateFields表达式。
|
UpdateFields用于在结构体中更新字段。
|
特殊的优化规则
| 规则 | 所属规则批 | 执行策略 | 解释说明 | 补充说明 |
|---|---|---|---|---|
| OptimizeMetadataOnlyQuery | Optimize Metadata Only Query | Once |
该规则优化了查询的执行，这些查询只能通过查看分区级别的元数据来回答。当扫描的所有列都是分区列，并且查询具有满足以下条件的聚合运算符：1. 聚合表达式是分区列。比如：SELECT col FROM tbl GROUP BY col 2. 聚合函数在分区列上使用了DISTINCT。比如：SELECT col1, count(DISTINCT col2) FROM tbl GROUP BY col1。3. 分区列上的聚合函数，这些列不管有没有DISTINCT关键字都具有相同的结果。比如：SELECT col1, Max(col2) FROM tbl GROUP BY col1
|
|
| PartitionPruning | PartitionPruning | Once |
根据JOIN操作的类型和选择执行动态分区裁剪（DPP）优化。在查询优化过程中，我们使用JOIN另一端的过滤器和名为DynamicPruning的自定义包装器在可过滤表上插入一个谓词。 DPP的基本机制是在满足以下条件时，从另一侧插入一个带有过滤器的重复子查询：1. 要修剪的表可通过JOIN KEY进行过滤 2. JOIN操作是以下类型之一：INNER、LEFT SEMI、LEFT OUTER（在右侧分区）或RIGHT OUTER（在左侧分区）为了在广播中直接启用分区裁剪，我们使用了一个定制的DynamicPruning子句，该子句将IN子句和子查询效益预估结合起来。在查询规划过程中，当JOIN类型已知时，我们使用以下机制：1. 如果连接是broadcast hash join，我们用广播的重用结果替换重复的子查询 2. 否则，如果分区裁剪的预估收益超过了两次运行子查询的开销，我们将保留重复的子查询 3. 否则，我们将删除子查询。
|
|
| PushDownPredicates | Pushdown Filters from PartitionPruning | fixedPoint |
常规的运算符和Join谓词下推的统一版本。此规则提高了级联join（例如：Filter-Join-Join-Join）的谓词下推性能。大多数谓词可以在一次传递中下推。
|
PushDownPredicates规则之前多次执行过。
|
| CleanupDynamicPruningFilters | Cleanup filters that cannot be pushed down | Once |
使用动态裁剪删除未下推到扫描的Filter节点。这些节点不会被下推到Project中，也不会使用非确定性表达式进行聚合。
|
|
| PruneFilters | Cleanup filters that cannot be pushed down | Once |
删除可以进行简单计算的Filter。这可以通过以下方式实现：1.在其计算结果始终为true的情况下，省略Filter。2.当筛选器的计算结果总是为false时，替换成一个伪空关系。3.消除子节点输出给定约束始终为true的条件。
|
该规则在前面的算子优化中执行过。
|
| ExtractPythonUDFFromJoinCondition | Extract Python UDFs | Once |
如果JOIN条件中的PythonUDF引用了JOIN双方的属性，则无法对其求值。有关详细信息，请参阅ExtractPythonUDFs。此规则将检测不可计算的PythonUDF，并将其从JOIN条件中拉出。
|
ExtractPythonUDFs用来从操作符中提取PythonUDFs，重写查询计划，以便可以在批处理中单独计算UDF。 仅提取可以在Python中计算的PythonUDFs（单个子级是PythonUDFs，或者所有子级都可以在JVM中计算）。 这有一个限制，即Python UDF的输入不允许包含来自多个子运算符的属性。
|
| CheckCartesianProducts | Extract Python UDFs | Once |
检查优化计划树中任何类型的join之间是否存在笛卡尔积。如果在未显示指定cross join的情况下找到笛卡尔积，则引发错误。如果CROSS_JOINS_ENABLED标志为true，则此规则将被有效禁用。 此规则必须在ReordJoin规则之后运行，因为在检查每个join是否为笛卡尔积之前，必须收集每个join的join条件。如果有SELECT * from R, S where R.r = S.s，则R和S之间的连接不是笛卡尔积，因此应该允许。谓词R.r=S.s在ReorderJoin规则之前不会被识别为join条件。 此规则必须在批处理LocalRelation之后运行，因为具有空关系的join不应是笛卡尔积。
|
ExtractPythonUDFFromJoinCondition规则会把JOIN转化成笛卡尔积，因此，我们需要重新运行笛卡尔积的检测。
|
| ExtractPythonUDFFromAggregate | Extract Python UDFs | Once |
提取逻辑聚合中的所有Python UDF，该UDF依赖于聚合表达式或分组键，或不依赖于上述任何表达式，在聚合后对其求值。
|
|
| ExtractGroupingPythonUDFFromAggregate | Extract Python UDFs | Once |
在逻辑聚合中提取PythonUDFs（分组键中使用），并在聚合之前对其求值。
|
这必须在ExtractPythonUDFFromAggregate之后和ExtractPythonUDFs之前执行。
|
| ExtractPythonUDFs | Extract Python UDFs | Once |
从运算符中提取PythonUDFs，重写查询计划，以便可以在批处理中单独计算UDF。 仅提取可以在Python中计算的PythonUDFs（单个子级是PythonUDFs，或者所有子级都可以在JVM中计算）。 这有一个限制，即Python UDF的输入不允许包含来自多个子运算符的属性。
|
|
| ColumnPruning | Extract Python UDFs | Once |
试图消除查询计划中不需要的列读取。 由于在Filter之前添加Project会和PushPredicatesThroughProject冲突，此规则将以以下模式删除Project p2： p1 @ Project(_, Filter(_, p2 @ Project(_, child))) if p2.outputSet.subsetOf(p2.inputSet) p2通常是按照这个规则插入的，没有用，p1无论如何都可以删减列。
|
eval-python节点可能位于ProjectFilter和scan节点之间，这会中断列裁剪和过滤器下推。这里我们重新运行相关的优化器规则。
|
| PushPredicateThroughNonJoin | Extract Python UDFs | Once |
当且仅当很多运算符满足下面条件时下推Filter运算符：1. 运算符是确定性的 2. 谓词是确定性的，运算符不会更改任何行。 假设表达式计算成本最小，这种启发式方法是有效的。
|
|
| RemoveNoopOperators | Extract Python UDFs | Once |
从查询计划中删除不进行任何修改的no-op运算符。
|
该规则在上面的算子优化和依赖统计信息的规则中都出现过。
|
| extraOptimizations | User Provided Optimizers | fixedPoint | 用户自定义的优化规则。 |
优化规则小结
Spark 3.3.0 版本共有 32 个规则批，172 个规则。
整体上分为标准的优化规则和特殊的优化规则，这是为了实现上的扩展性。
标准的优化规则分为算子优化前、算子优化和依赖统计数据的优化。
算子优化前主要涉及 analysis 阶段的一点收尾规则、Union、Limit、数据库关系、子查询、算子的替代、聚合算子。
算子优化包括 4 个阶段，分为过滤推断前、中、后以及JOIN谓词下推。
其中过滤推断前后的优化规则都是一样的，主要是因为过滤推断会对算子的优化产生影响。
过滤推断前的算子优化主要包括 3 个方面的优化——算子下推、算子合并以及常量折叠/强度消减。
算子下推主要涉及到Project、Join、Limit、列剪裁、Generate。
算子合并主要涉及Repartition、Project、Window、Filter、Limit、Union。
常量折叠/强度消减主要涉及Repartition、Window、Null、常量、In、Filter、整数类型、Like、Boolean、if/case、二义性、no-op、struct、取值操作（struct/array/map）、csv/json、Concat。
过滤推断就是单纯关于Filter算子的优化。
JOIN谓词下推明显和是和JOIN谓词相关。
依赖统计数据的优化规则涉及Project、Filter、Join、Sort、Decimal、Aggregate、对象表达式、数据库关系、笛卡尔积、子查询、Float、Struct。
这里值得注意的是依赖统计数据的优化规则中还包含一个特殊的 CBO（Cost Based Optimization，基于成本的优化） 优化规则，即CostBasedJoinReorder，它可以基于成本的优化规则来重排序 JOIN。
特殊的优化规则涉及分区元数据、DPP（动态分区裁剪）、Filter、Python UDF以及用户自定义的优化规则。
特殊的优化规则中最值得注意的就是动态分区裁剪 DPP。
回到最初的例子
和上一讲的结构类似，我们需要重点针对几个优化规则详解。
参考上一讲，我们配置好spark.sql.planChangeLog.level后，运行程序。然后看看控制台日志打印，到底有哪些优化规则参与到了我们例子的优化当中？
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Eliminate Distinct has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger:
=== Applying Rule org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases ===GlobalLimit 21 GlobalLimit 21+- LocalLimit 21 +- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11] +- Project [addr#8, age#9L, name#10, sex#11]
! +- SubqueryAlias t_user +- View (`t_user`, [addr#8,age#9L,name#10,sex#11])
! +- View (`t_user`, [addr#8,age#9L,name#10,sex#11]) +- Relation [addr#8,age#9L,name#10,sex#11] json
! +- Relation [addr#8,age#9L,name#10,sex#11] json 22/03/29 21:32:22 INFO PlanChangeLogger:
=== Applying Rule org.apache.spark.sql.catalyst.analysis.EliminateView ===GlobalLimit 21 GlobalLimit 21+- LocalLimit 21 +- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11] +- Project [addr#8, age#9L, name#10, sex#11]
! +- View (`t_user`, [addr#8,age#9L,name#10,sex#11]) +- Relation [addr#8,age#9L,name#10,sex#11] json
! +- Relation [addr#8,age#9L,name#10,sex#11] json 22/03/29 21:32:22 INFO PlanChangeLogger:
=== Result of Batch Finish Analysis ===GlobalLimit 21 GlobalLimit 21+- LocalLimit 21 +- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11] +- Project [addr#8, age#9L, name#10, sex#11]
! +- SubqueryAlias t_user +- Relation [addr#8,age#9L,name#10,sex#11] json
! +- View (`t_user`, [addr#8,age#9L,name#10,sex#11])
! +- Relation [addr#8,age#9L,name#10,sex#11] json 22/03/29 21:32:22 INFO PlanChangeLogger:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.RemoveNoopOperators ===GlobalLimit 21 GlobalLimit 21+- LocalLimit 21 +- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]
! +- Project [addr#8, age#9L, name#10, sex#11] +- Relation [addr#8,age#9L,name#10,sex#11] json
! +- Relation [addr#8,age#9L,name#10,sex#11] json 22/03/29 21:32:22 INFO PlanChangeLogger:
=== Result of Batch Union ===GlobalLimit 21 GlobalLimit 21+- LocalLimit 21 +- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]
! +- Project [addr#8, age#9L, name#10, sex#11] +- Relation [addr#8,age#9L,name#10,sex#11] json
! +- Relation [addr#8,age#9L,name#10,sex#11] json 22/03/29 21:32:22 INFO PlanChangeLogger: Batch OptimizeLimitZero has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch LocalRelation early has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Pullup Correlated Expressions has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Subquery has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Replace Operators has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Aggregate has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.SimplifyCasts ===GlobalLimit 21 GlobalLimit 21+- LocalLimit 21 +- LocalLimit 21
! +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, name#10 AS name#26, sex#11 AS sex#27]+- Relation [addr#8,age#9L,name#10,sex#11] json +- Relation [addr#8,age#9L,name#10,sex#11] json22/03/29 21:32:22 INFO PlanChangeLogger:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.RemoveRedundantAliases ===GlobalLimit 21 GlobalLimit 21+- LocalLimit 21 +- LocalLimit 21
! +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, name#10 AS name#26, sex#11 AS sex#27] +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, name#10, sex#11]+- Relation [addr#8,age#9L,name#10,sex#11] json +- Relation [addr#8,age#9L,name#10,sex#11] json22/03/29 21:32:22 INFO PlanChangeLogger:
=== Result of Batch Operator Optimization before Inferring Filters ===GlobalLimit 21 GlobalLimit 21+- LocalLimit 21 +- LocalLimit 21
! +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, name#10, sex#11]+- Relation [addr#8,age#9L,name#10,sex#11] json +- Relation [addr#8,age#9L,name#10,sex#11] json22/03/29 21:32:22 INFO PlanChangeLogger: Batch Infer Filters has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Operator Optimization after Inferring Filters has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Push extra predicate through join has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Early Filter and Projection Push-Down has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Update CTE Relation Stats has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Join Reorder has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Eliminate Sorts has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Decimal Optimizations has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Distinct Aggregate Rewrite has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Object Expressions Optimization has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch LocalRelation has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Optimize One Row Plan has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Check Cartesian Products has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch RewriteSubquery has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch NormalizeFloatingNumbers has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch ReplaceUpdateFieldsExpression has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Optimize Metadata Only Query has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch PartitionPruning has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Pushdown Filters from PartitionPruning has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Cleanup filters that cannot be pushed down has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch Extract Python UDFs has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger: Batch User Provided Optimizers has no effect.
22/03/29 21:32:22 INFO PlanChangeLogger:
=== Metrics of Executed Rules ===
Total number of runs: 221
Total time: 0.034200906 seconds
Total number of effective runs: 5
Total time of effective runs: 0.005719922 seconds
从上面的日志打印中可以得到我们的目标：
- EliminateSubqueryAliases
- EliminateView
- RemoveNoopOperators
- SimplifyCasts
- RemoveRedundantAliases
我们来一一详解~
EliminateSubqueryAliases
WHAT
EliminateSubqueryAliases 用来消除子查询别名，对应逻辑算子树中的SubqueryAlias节点。一般来讲，Subqueries 仅用于提供查询的视角范围(Scope)信息，一旦 analysis 阶段结束， 该节点就可以被移除，该优化规则直接将SubqueryAlias替换为其子节点。
HOW
那么它是如何实现的呢？
照例，先看看对应规则的apply方法源码：
def apply(plan: LogicalPlan): LogicalPlan =
// 闭包外层用来记录 resolveOperator 调用的深度
AnalysisHelper.allowInvokingTransformsInAnalyzer {plan.transformUpWithPruning(AlwaysProcess.fn, ruleId) {case SubqueryAlias(_, child) => child}}
plan.transformUpWithPruning 的源码详情请参考上一讲，它会返回当前节点的副本，其中的规则首先递归应用于所有的子节点，然后递归应用于自身（后序）。当规则不适用于给定节点时，它会保持不变。
case SubqueryAlias(_, child) => child
这行代码就很好理解了，如果当前的节点是SubqueryAlias类型的，只保留它的子节点就行了。
变化
从控制台的日志中就可以发现，逻辑计划变化前：
GlobalLimit 21 +- LocalLimit 21 +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [addr#8, age#9L, name#10, sex#11] +- SubqueryAlias t_user +- View (`t_user`, [addr#8,age#9L,name#10,sex#11]) +- Relation [addr#8,age#9L,name#10,sex#11] json
逻辑计划变化后：
GlobalLimit 21+- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11]+- View (`t_user`, [addr#8,age#9L,name#10,sex#11])+- Relation [addr#8,age#9L,name#10,sex#11] json
是不是只是简单的脱去了SubqueryAlias 的外壳，只保留了其子节点？
EliminateView
WHAT
EliminateView会从计划中删除View算子。在analysis阶段结束之前，这个算子会一直得到尊重，因为我们希望看到AnalyzedLogicalPlan的哪一部分是从视图生成的。
HOW
我们看看它是怎么实现的？
override def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {case View(_, _, child) => child}
transformUp 底层调用的就是 transformUpWithPruning ，都是通过后序遍历的方式递归地将规则应用于整棵树。
case View(_, _, child) => child
这行代码和上个规则类似，都是脱去外壳，只保留子节点。
变化
逻辑计划变化前：
GlobalLimit 21 +- LocalLimit 21 +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [addr#8, age#9L, name#10, sex#11] +- View (`t_user`, [addr#8,age#9L,name#10,sex#11]) +- Relation [addr#8,age#9L,name#10,sex#11] json
逻辑计划变化后：
GlobalLimit 21+- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]+- Project [addr#8, age#9L, name#10, sex#11]+- Relation [addr#8,age#9L,name#10,sex#11] json
上面的变化完全符合我们对EliminateView这个优化规则的预期。
RemoveNoopOperators
WHAT
RemoveNoopOperators用来从查询计划中删除不进行任何修改的no-op运算符。
HOW
我们看看它是怎么实现的~
def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(_.containsAnyPattern(PROJECT, WINDOW), ruleId) {// 消除 no-op 的 Projectscase p @ Project(projectList, child) if child.sameOutput(p) =>val newChild = child match {case p: Project =>p.copy(projectList = restoreOriginalOutputNames(p.projectList, projectList.map(_.name)))case agg: Aggregate =>agg.copy(aggregateExpressions =restoreOriginalOutputNames(agg.aggregateExpressions, projectList.map(_.name)))case _ =>child}if (newChild.output.zip(projectList).forall { case (a1, a2) => a1.name == a2.name }) {newChild} else {p}// 消除 no-op 的 Windowcase w: Window if w.windowExpressions.isEmpty => w.child}
child.sameOutput(p)这个方法当且仅当child和p的输出（output）在语义上相同，则返回true，即：
- 它包含相同数量的属性（Attribute）；
- 引用是相同的；
- 顺序也是一样的。
上面的代码也很好理解，no-op 就是无操作的意思， 碰到Project和Window算子，但是外面这层壳没啥子作用，那我们脱去这层壳就行了。
和前面 2 个优化规则都是类似的，只不过这里有个限制条件是no-op，而前面 2 个优化规则可不管三七二十一，直接就脱。
变化
逻辑计划变化前：
GlobalLimit 21 +- LocalLimit 21 +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Project [addr#8, age#9L, name#10, sex#11] +- Relation [addr#8,age#9L,name#10,sex#11] json
逻辑计划变化后：
GlobalLimit 21
+- LocalLimit 21
+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27]
+- Relation [addr#8,age#9L,name#10,sex#11] json
细心的同学可能问了：逻辑计划变化前明明有 2 个 Project，为啥逻辑计划变化后还剩下一个呢？
因为剩下的那个Project不是no-op啊，你看里面是不是有个cast操作。
SimplifyCasts
WHAT
SimplifyCasts用来删除不必要的强制转换，因为输入已经是正确的类型。
HOW
def apply(plan: LogicalPlan): LogicalPlan = plan.transformAllExpressionsWithPruning(_.containsPattern(CAST), ruleId) {// 当前节点和子节点的数据类型相同，只保留子节点case Cast(e, dataType, _, _) if e.dataType == dataType => e// 存在 cast 嵌套的情况，判断 2 种数值类型是否兼容，即能不能强转过去case c @ Cast(Cast(e, dt1: NumericType, _, _), dt2: NumericType, _, _)if isWiderCast(e.dataType, dt1) && isWiderCast(dt1, dt2) =>// 复制一份，保留子节点c.copy(child = e)case c @ Cast(e, dataType, _, _) => (e.dataType, dataType) match {// 如果是数组类型，第二个参数表示是否可以包含空值case (ArrayType(from, false), ArrayType(to, true)) if from == to => e// 如果是 Map 类型，并且键值的类型都是一致的case (MapType(fromKey, fromValue, false), MapType(toKey, toValue, true))if fromKey == toKey && fromValue == toValue => ecase _ => c}}
变化
逻辑计划变化前：
GlobalLimit 21 +- LocalLimit 21 +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, cast(name#10 as string) AS name#26, cast(sex#11 as string) AS sex#27] +- Relation [addr#8,age#9L,name#10,sex#11] json
逻辑计划变化后：
GlobalLimit 21
+- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, name#10 AS name#26, sex#11 AS sex#27]+- Relation [addr#8,age#9L,name#10,sex#11] json
参考上一讲，我们从 JSON 文件中已经推测出了字段对应的类型。
| 字段名称 | 字段类型 |
|---|---|
| name | StringType |
| age | LongType |
| sex | StringType |
| addr | ArrayType |
name 和 sex 都已经是字符串类型了，故没有必要进行强制转换。
优化规则的作用完全符合我们的预期。
RemoveRedundantAliases
WHAT
RemoveRedundantAliases用来从查询计划中删除冗余别名。
冗余别名是不会更改列的名称或元数据，也不会消除重复数据的别名。
HOW
def apply(plan: LogicalPlan): LogicalPlan = removeRedundantAliases(plan, AttributeSet.empty)/*** 从逻辑计划及其子树中删除冗余的别名表达式。* `excluded`用于防止删除看似冗余的别名，这些别名用于消除（self）join 的输入的重复，或防止删除顶层的子查询属性。*/private def removeRedundantAliases(plan: LogicalPlan, excluded: AttributeSet): LogicalPlan = {// 没有包含别名直接返回if (!plan.containsPattern(ALIAS)) {return plan}plan match {// 我们希望子查询保持相同的输出属性。这意味着我们不能移除产生这些属性的别名case Subquery(child, correlated) =>Subquery(removeRedundantAliases(child, excluded ++ child.outputSet), correlated)// `JOIN`必须以不同的方式处理，因为`JOIN`的左侧和右侧不允许使用相同的属性。我们使用排除列表来阻止我们发生这种情况；只有当别名的子节点属性不在黑名单上规则才会删除别名。case Join(left, right, joinType, condition, hint) =>val newLeft = removeRedundantAliases(left, excluded ++ right.outputSet)val newRight = removeRedundantAliases(right, excluded ++ newLeft.outputSet)val mapping = AttributeMap(createAttributeMapping(left, newLeft) ++createAttributeMapping(right, newRight))val newCondition = condition.map(_.transform {case a: Attribute => mapping.getOrElse(a, a)})Join(newLeft, newRight, joinType, newCondition, hint)case _ =>// 移除子树中的冗余别名val currentNextAttrPairs = mutable.Buffer.empty[(Attribute, Attribute)]val newNode = plan.mapChildren { child =>val newChild = removeRedundantAliases(child, excluded)currentNextAttrPairs ++= createAttributeMapping(child, newChild)newChild}// 创建属性映射。请注意，`currentNextAttrPairs`可以包含重复的`Union`情况下的键（这是由`PushProjectionThroughUnion`规则引起的）；在这种情况下，我们使用第一个映射（应该由第一个子节点提供）。val mapping = AttributeMap(currentNextAttrPairs.toSeq)// 为实际可能产生冗余别名的节点创建表达式清理函数，否则使用`identity`。val clean: Expression => Expression = plan match {case _: Project => removeRedundantAlias(_, excluded)case _: Aggregate => removeRedundantAlias(_, excluded)case _: Window => removeRedundantAlias(_, excluded)case _ => identity[Expression]}// 转换表达式newNode.mapExpressions { expr =>clean(expr.transform {case a: Attribute => mapping.get(a).map(_.withName(a.name)).getOrElse(a)})}}}/*** 移除一个表达式顶层冗余的别名*/private def removeRedundantAlias(e: Expression, excludeList: AttributeSet): Expression = e match {// 无法剥离带有元数据的别名，否则元数据将丢失。// 如果别名与属性名不同，我们也不能删除它，否则可能会意外更改根计划的输出`schema`名称。case a @ Alias(attr: Attribute, name)if (a.metadata == Metadata.empty || a.metadata == attr.metadata) &&name == attr.name &&!excludeList.contains(attr) &&!excludeList.contains(a) =>attrcase a => a}
变化
逻辑计划变化前：
GlobalLimit 21
+- LocalLimit 21 +- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, name#10 AS name#26, sex#11 AS sex#27] +- Relation [addr#8,age#9L,name#10,sex#11] json
逻辑计划变化后：
GlobalLimit 21
+- LocalLimit 21+- Project [cast(addr#8 as string) AS addr#24, cast(age#9L as string) AS age#25, name#10, sex#11]+- Relation [addr#8,age#9L,name#10,sex#11] json
对比后很容易发现，name 和 sex 这两个字段的别名没有存在的必要，故这个优化规则就把它们都删除了。
最终优化后的逻辑计划
照例，我们还是画一个表格将上面的逻辑计划和具体的源码对应起来：
| 打印 | 对应的源码 | 说明 |
|---|---|---|
| GlobalLimit | org.apache.spark.sql.catalyst.plans.logical.GlobalLimit | 全局的 Limit |
| 21 | GlobalLimit.maxRows | 最多返回的数据行 |
| LocalLimit | org.apache.spark.sql.catalyst.plans.logical.LocalLimit | 分区局部的 Limit |
| 21 | LocalLimit.maxRowsPerPartition | 单个分区最多返回的数据行 |
| Project | org.apache.spark.sql.catalyst.plans.logical.Project | 投影对象 |
| cast(addr#8 as string) AS addr#24 | org.apache.spark.sql.catalyst.expressions.Alias | 别名对象，用来给一个计算分配一个新的名称，数字代表的表达式 ID |
| cast(age#9L as string) AS age#25 | org.apache.spark.sql.catalyst.expressions.Alias | 别名对象，用来给一个计算分配一个新的名称，数字代表的表达式 ID |
| name#10 | org.apache.spark.sql.catalyst.expressions.AttributeReference | 对树中另一个操作符生成的属性的引用，数字代表的表达式 ID |
| sex#11 | org.apache.spark.sql.catalyst.expressions.AttributeReference | 对树中另一个操作符生成的属性的引用，数字代表的表达式 ID |
| Relation [addr#8,age#9L,name#10,sex#11] json | LogicalRelation.simpleString | 属性序列和关系名称，数字代表的表达式 ID |
总结
本讲是对 Spark SQL 工作流程中 optimization 阶段的源码解析。
我们先明确了自己的目标——我们要规范化逻辑计划，以便充分利用缓存来加速查询，同时我们要利用各种 SQL 优化手段，使得我们的逻辑计划树更好更方便执行。
然后我们从 Dataset.show 方法开始，一步步找到optimization阶段的入口，在这个过程中我们对 Dataset.show 的底层实现原理有了充分的认识。
正式开始optimization阶段之后，我们的逻辑计划先从CacheManager中走了一遭。
在这个过程中，我们规范化了逻辑计划树，缓存了查询结果，并在执行后续查询时自动使用这些缓存结果。
接着我们梳理了所有的优化规则，优化规则分为 2 大类：标准的优化规则和特殊的优化规则。
其中标准的优化规则分为算子优化前、算子优化和依赖统计数据的优化。
算子优化包括 4 个阶段，分为过滤推断前、中、后以及JOIN谓词下推。
依赖统计数据的优化中包含一个特殊的 CBO 优化规则，即基于成本来优化 JOIN 排序。
特殊的优化规则中最值得注意的就是动态分区裁剪 DPP。
和上一讲的结构类似，由于篇幅的限制，我们不可能做到详解每一种优化规则具体是怎么实现的。
所以，我们就回到了我们最初的例子，通过配置了spark.sql.planChangeLog.level后的控制台打印，我们知道有 5 个优化规则参与了例子的优化。
- EliminateSubqueryAliases
- EliminateView
- RemoveNoopOperators
- SimplifyCasts
- RemoveRedundantAliases
我们详细解析了这 5 个规则是如何作用于我们的逻辑计划，最终完成 optimization 阶段的优化的。
至此，optimization 阶段的基本流程相信大家已经一览无余了。
麻烦看到这里的同学帮忙三连支持一波，万分感谢~
Spark SQL 工作流程源码解析（四）optimization 阶段（基于 Spark 3.3.0）相关推荐
- Spark ALS recommendForAll源码解析实战之Spark1.x vs Spark2.x
文章目录 Spark ALS recommendForAll源码解析实战 1. 软件版本: 2. 本文要解决的问题 3. 源码分析实战 3.1 Spark2.2.2 ALS recommendForA ...
- 云原生小课堂｜Envoy请求流程源码解析（三）：请求解析
前言 Envoy 是一款面向 Service Mesh 的高性能网络代理服务.它与应用程序并行运行,通过以平台无关的方式提供通用功能来抽象网络.当基础架构中的所有服务流量都通过 Envoy 网格时 ...
- Quantopian教程源码解析及实战
Quantopian教程源码解析及实战 本文主要侧重于采用Quantopian进行实际的金融策略分析,因此阅读本文前,需要对Quantopian的有基本的了解,建议先阅读Quantopian的tu ...
- Android Launcher启动应用程序流程源码解析
带着问题看源码 点击桌面Launcher图标后做了哪些工作? 应用程序什么时候被创建的? Application和MainActivity的onCreate()方法什么时候被调用的? 概述 在Andr ...
- Myth源码解析系列之六- 订单下单流程源码解析（发起者）
前面一章我们走完了服务启动的源码,这次我们进入下单流程的源码解析~ 订单下单流程源码解析(发起者) 首先保证myth-demo-springcloud-order.myth-demo-springcl ...
- swoole 启动流程_EasySwoole 服务启动过程以及主体设计流程源码解析
EasySwoole 服务启动过程以及主体设计流程源码解析 本文主要讲解EasySwoole 服务的启动过程,会通过源码片段讲解主体的设计流程 命令启动 当我们通过php easyswoole sta ...
- 快学Big Data -- Spark SQL总结（二十四)
Spark SQL 总结 概述 Spark Sql 是用来处理结构化数据的一个模块,它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用. 特点 spark sql 要比 ...
- spark sql 本地调试_干货 | 如何成为大数据Spark高手
Spark是发源于美国加州大学伯克利分校AMPLab的集群计算平台,它立足于内存计算,性能超过Hadoop百倍,从多迭代批量处理出发,兼收并蓄数据仓库.流处理和图计算等多种计算范式,是罕见的全能选手. ...
- SpringBoot启动全流程源码解析（超详细版）
我们在使用SpringBoot启动项目的时候,可能只需加一个注解,然后启动main,整个项目就运行了起来,但事实真的是所见即所得吗,还是SpringBoot在背后默默做了很多?本文会通过源码解析的方式 ...
最新文章
- Linux软连接和硬链接
- office2003/2007/2010版本降低宏安全设置方法
- IDC中国AI服务器报告出炉：浪潮继续蝉联第一，份额超过其他玩家总和
- java 从数据库取值_JAVA操作数据库--从一张表中取值，经过判断，然后插入另一张表中。...
- mysql timezone utc_在与mysql的配置当中一定要注意的一点是设置serverTimezone=UTC
- disruptor入门
- torch.nn.Module()
- 海归硕士程序员吐槽：我回国竟然找不到工作，这到底是什么操作？
- Hie with the Pie（poj3311）
- ThreadLocal学习
- 度身定造的女孩子C程序
- 基于 Vue 的轻量级静态网站生成器 VuePress
- 利用计算机系统辅助设计人员,计算机基础知识：计算机辅助设计
- AS3 库资源 很多非常有用的类库
- 世园会开幕式上的机器人_小胖机器人驻扎“2019北京世园会”媒体分会场
- 读书笔记-opencv-极坐标变换
- Python3入门机器学习之2.8scikit-learn中的Scaler
- windows ESP分区丢失处理方案与磁盘分区注意事项
- 内网安全检测技术yyds！你还不会吗？
- 力扣 6181. 最长的字母序连续子字符串的长度