Spark优化篇：Explain

本章，主要针对Spark3.x版本后Explain去进行详细的介绍，Explain也是企业生产调优的一个重要方式、策略。执行计划流程图：执行计划阶段详解：阶段阶段类型解释1Unresolved Logical Plan检查 SQL 语法上是否有问题，然后生成 Unresolved（未决断）的逻辑计划，2Logical Plan通过访问 Spark 中的 Catalog 存储库来解析验证语义、列名、

YaPengLi.

1553人浏览 · 2022-04-14 18:04:55

YaPengLi. · 2022-04-14 18:04:55 发布

本章，主要针对Spark3.x版本后Explain去进行详细的介绍，Explain也是企业生产调优的一个重要方式、策略。

执行计划流程图：

执行计划阶段详解：

阶段	阶段类型	解释
1	Unresolved Logical Plan	检查 SQL 语法上是否有问题，然后生成 Unresolved（未决断）的逻辑计划，
2	Logical Plan	通过访问 Spark 中的 Catalog 存储库来解析验证语义、列名、类型、表名等。
3	Optimized Logical Plan	Catalyst 优化器根据RBO各种规则进行优化。
4	Physical Plan	优化后的逻辑执行计划转化为物理执行计划。
5	Cost Physical Plan	根据合适CBO的Cost（成本模型）将物理执行计划转化为可以执行的代码。
6	RDDS	生成最终执行的RDD。

Explain 参数详解：

explain(mode="simple")

序号	阶段类型	解释
1	simple	只展示物理执行计划。
2	extended	展示物理执行计划和逻辑执行计划。
3	codegen	展示要 Codegen 生成的可执行 Java 代码。
4	cost	展示优化后的逻辑执行计划以及相关的统计。
5	formatted	以分隔的方式输出，它会输出更易读的物理执行计划，并展示每个节点的详细信息。

执行计划关键字详解：

序号	阶段类型	解释
1	HashAggregate	运算符表示数据聚合，一般 HashAggregate 是成对出现，第一个HashAggregate 是将执行节点本地的数据进行局部聚合，另一个 HashAggregate 是将各个分区的数据进一步进行聚合计算。
2	Exchange	代表Shuffle，表示需要在集群上移动数据。很多时候HashAggregate 会以 Exchange 分隔开来。
3	Project	SQL 中的裁剪操作，就是列选择。如：select name, id…
4	BroadcastHashJoin	表示通过基于广播方式进行 Hash Join。
5	LocalTableScan	运算符就是全表扫描本地的表。

测试准备：

    var appSql: String =
      """
        |select
        |   t1.name,count(1)
        |from
        |   tab_spark_test as t1
        |left join tab_spark_test_2 as t2
        |on t1.id = t2.id
        |group by t1.name
        """.stripMargin

sparkSession.sql(appSql).explain(mode = "extended")

分别为,未决断执行计划->决断后执行计划->逻辑执行计划->CBO执行计划

== Parsed Logical Plan ==
GlobalLimit 21
+- LocalLimit 21
   +- Project [cast(name#2 as string) AS name#10, cast(count(1)#5L as string) AS count(1)#11]
      +- Aggregate [name#2], [name#2, count(1) AS count(1)#5L]
         +- Join LeftOuter, (id#1 = id#3)
            :- SubqueryAlias t1
            :  +- SubqueryAlias spark_catalog.default.tab_spark_test
            :     +- HiveTableRelation `default`.`tab_spark_test`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#1, name#2]
            +- SubqueryAlias t2
               +- SubqueryAlias spark_catalog.default.tab_spark_test_2
                  +- HiveTableRelation `default`.`tab_spark_test_2`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#3, name#4]

== Analyzed Logical Plan ==
name: string, count(1): string
GlobalLimit 21
+- LocalLimit 21
   +- Project [cast(name#2 as string) AS name#10, cast(count(1)#5L as string) AS count(1)#11]
      +- Aggregate [name#2], [name#2, count(1) AS count(1)#5L]
         +- Join LeftOuter, (id#1 = id#3)
            :- SubqueryAlias t1
            :  +- SubqueryAlias spark_catalog.default.tab_spark_test
            :     +- HiveTableRelation `default`.`tab_spark_test`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#1, name#2]
            +- SubqueryAlias t2
               +- SubqueryAlias spark_catalog.default.tab_spark_test_2
                  +- HiveTableRelation `default`.`tab_spark_test_2`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#3, name#4]

== Optimized Logical Plan ==
GlobalLimit 21
+- LocalLimit 21
   +- Aggregate [name#2], [name#2, cast(count(1) as string) AS count(1)#11]
      +- Project [name#2]
         +- Join LeftOuter, (id#1 = id#3)
            :- HiveTableRelation `default`.`tab_spark_test`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#1, name#2]
            +- Project [id#3]
               +- Filter isnotnull(id#3)
                  +- HiveTableRelation `default`.`tab_spark_test_2`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#3, name#4]

== Physical Plan ==
CollectLimit 21
+- *(3) HashAggregate(keys=[name#2], functions=[count(1)], output=[name#2, count(1)#11])
   +- Exchange hashpartitioning(name#2, 200), true, [id=#48]
      +- *(2) HashAggregate(keys=[name#2], functions=[partial_count(1)], output=[name#2, count#15L])
         +- *(2) Project [name#2]
            +- *(2) BroadcastHashJoin [id#1], [id#3], LeftOuter, BuildRight
               :- Scan hive default.tab_spark_test [id#1, name#2], HiveTableRelation `default`.`tab_spark_test`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#1, name#2]
               +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint))), [id=#42]
                  +- *(1) Filter isnotnull(id#3)
                     +- Scan hive default.tab_spark_test_2 [id#3], HiveTableRelation `default`.`tab_spark_test_2`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#3, name#4]