Spark性能调优：执行计划详解

Spark性能调优：执行计划详解

引用

CSDN

1.

https://blog.csdn.net/weixin_41914554/article/details/145463095

在Spark中，explain函数用于提供数据框（DataFrame）或SQL查询的逻辑计划和物理执行计划的详细解释。它可以帮助开发者理解Spark是如何执行查询的，包括优化过程、转换步骤以及它将采用的物理执行策略。

逻辑计划 (Logical Plan)

逻辑计划代表了Spark将应用于处理数据的抽象操作序列。它是基于用户提供的DataFrame API或SQL查询，经过优化前的中间表示。

• 未优化的逻辑计划：这是查询的初始形式，尚未应用任何优化。
• 优化后的逻辑计划：这是经过Spark应用了一些规则（如谓词下推、常量折叠、简化等）后的查询计划。

物理计划 (Physical Plan)

物理计划表示Spark如何实际执行查询。它包括如何对数据进行Shuffle（洗牌）、分区或合并的详细信息。物理计划通常有多个选择，Spark会根据成本（如计算开销、数据传输等）选择最优的执行计划。

• RDD（弹性分布式数据集）操作：物理计划会显示Spark执行数据处理时所使用的具体转换步骤和操作。

基于成本的优化 (Cost-Based Optimization)

在某些情况下，Spark还可以执行基于成本的优化（如选择不同的连接策略或决定是否执行广播连接），具体取决于可用的统计信息。

使用示例

PySpark 示例：

df = spark.read.csv("data.csv", header=True, inferSchema=True)
df_filtered = df.filter(df["age"] > 25).select("name", "age")
df_filtered.explain(True)

Spark SQL 示例：

spark.sql("SELECT name, age FROM people WHERE age > 25").explain(True)

输出 explain(True) 的结果：

当调用 explain(True) 时，您会看到一个包含逻辑计划和物理计划的详细输出。以下是一个简单的输出示例：

   +- *Scan ExistingRDD[name#0, age#1] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/data.csv], PartitionFilters: [], PushedFilters: [GreaterThan(age,25)], ReadSchema: struct<name:string,age:int>
   +- Relation[name#0, age#1] csv

输出内容的关键元素：

1. 物理计划：

• Project：表示选取了哪些列。
• Filter：表示应用的过滤条件。
• Scan ExistingRDD：表示数据的读取来源，在这个例子中是CSV文件。

2. 逻辑计划：

• 逻辑计划中显示了用户代码中定义的Project和Filter操作。

使用 explain 的场景：

• 优化调试：可以使用explain来检查Spark是否正确优化了查询。
• 连接优化：如果您不确定Spark如何处理连接，explain可以帮助您查看Spark是否使用了合适的连接策略（例如，广播连接或Shuffle连接）。
• 性能调优：通过检查物理计划，您可以识别出Spark可能执行的无谓操作，或者有机会进一步优化的地方。

详细语法

EXPLAIN [ EXTENDED | CODEGEN | COST | FORMATTED ] statement

EXTENDED
生成解析后的逻辑计划、分析后的逻辑计划、优化后的逻辑计划和物理计划。
解析后的逻辑计划是从查询中提取的未解析计划。
分析后的逻辑计划通过将未解析的属性（unresolvedAttribute）和未解析的关系（unresolvedRelation）转换为完全类型化的对象来完成转换。
优化后的逻辑计划通过一组优化规则进行转换，最终生成物理计划。

CODEGEN
生成语句的代码（如果有）和物理计划。

COST
如果计划节点的统计信息可用，生成逻辑计划和统计信息。

FORMATTED
生成两个部分：物理计划概述和节点详细信息。

statement
指定要解释的SQL语句。

EXPLAIN select k, sum(v) from values (1, 2), (1, 3) t(k, v) group by k;
+---+
| == Physical Plan ==
*(2) HashAggregate(keys=[k#33], functions=[sum(cast(v#34 as bigint))])
+- Exchange hashpartitioning(k#33, 200), true, [id=#59]
   +- *(1) HashAggregate(keys=[k#33], functions=[partial_sum(cast(v#34 as bigint))])
      +- *(1) LocalTableScan [k#33, v#34]
+---+
EXPLAIN EXTENDED select k, sum(v) from values (1, 2), (1, 3) t(k, v) group by k;
+---+
| == Parsed Logical Plan ==
'Aggregate ['k], ['k, unresolvedalias('sum('v), None)]
    +- 'UnresolvedInlineTable [k, v], [List(1, 2), List(1, 3)]
== Analyzed Logical Plan ==
Aggregate [k#47], [k#47, sum(cast(v#48 as bigint)) AS sum(v)#50L]
    +- LocalRelation [k#47, v#48]
== Optimized Logical Plan ==
Aggregate [k#47], [k#47, sum(cast(v#48 as bigint)) AS sum(v)#50L]
+- LocalRelation [k#47, v#48]
*(2) HashAggregate(keys=[k#47], functions=[sum(cast(v#48 as bigint))], output=[k#47, sum(v)#50L])
+- Exchange hashpartitioning(k#47, 200), true, [id=#79]
   +- *(1) HashAggregate(keys=[k#47], functions=[partial_sum(cast(v#48 as bigint))], output=[k#47, sum#52L])
      +- *(1) LocalTableScan [k#47, v#48]
+---+
EXPLAIN FORMATTED select k, sum(v) from values (1, 2), (1, 3) t(k, v) group by k;
+---+
(1) LocalTableScan [codegen id : 1]
(2) HashAggregate [codegen id : 1]
(4) HashAggregate [codegen id : 2]