sparkSQL执行计划详解

灰太狼 2022-12-13 01:30 497阅读 0赞

一：首先将我们从文本中读取的数据映射到表也就是视图

eg:

$>cat b.txt

1 ded  
　　2 dsfre  
　　3 sfs  
　　4 fr

$>val sc = spark.sparkContext 　　　　\#创建SparkContext

$>val rdd = sc.textFile("file:///home/Alex\_lei/b.txt").map(x=>x.split(" ")).map(x=>(x(0),x(1)))

\#读取文件到rdd中（tuple形式，因为createDataFrame方法所需要的rdd为tuple形式）

$>val df = spark.createDataFrame(rdd)　　　　\#创建dataframe

$>df.createTempView("person")　　　　　　　　\#将dataframe映射到表

二：分析

$>val query = spark.sql("select \* from person where \_1>1")

（1）explain() 查看物理计划

$>query.explain()

== Physical Plan ==

\*Filter (isnotnull(\_1\#3) && (cast(\_1\#3 as double) > 1.0))  
　　　　　　　　+- \*SerializeFromObject \[staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, 　　　　　　　　　　assertnotnull(input\[0, scala.Tuple2, true\]).\_1, true) AS \_1\#3, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input\[0, scala.Tuple2, true\]).\_2, true) AS \_2\#4\]  
　　　　　　　　　　+- Scan ExternalRDDScan\[obj\#2\]

说明：类似一棵树，从下往上看，首先扫描外部RDD，然后是序列化字段，在就是过滤，判断是否为null和第一个字段大于1的。

（2）explain(true)查看整个SQL的执行计划，主要分为4个阶段

--1:解析过程

== Parsed Logical Plan ==

'Project \[\*\]  
　　　　　　+- 'Filter ('\_1 > 1)  
　　　　　　　　+- 'UnresolvedRelation \`person\`

说明：Project\[\*\]是我们所要的结果集，解析过程不能判断表person是否存在，有什么关系，然后就是列出过滤条件和所要的结果集。

--2:逻辑阶段

== Analyzed Logical Plan ==

\_1: string, \_2: string  
　　　　　　Project \[\_1\#3, \_2\#4\]  
　　　　　　　　+- Filter (cast(\_1\#3 as double) > cast(1 as double))  
　　　　　　　　　　+- SubqueryAlias person  
　　　　　　　　　　　　+- SerializeFromObject \[staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input\[0, scala.Tuple2, true\]).\_1, true) AS \_1\#3, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input\[0, scala.Tuple2, true\]).\_2, true) AS \_2\#4\]  
　　　　　　　　　　　　　　+- ExternalRDD \[obj\#2\]

说明：首先还是加载外部RDD，然后序列化字段，列出映射表的名字，确认表存在，然后按照条件过滤，获取结果集。

--3:优化阶段

== Optimized Logical Plan ==

Filter (isnotnull(\_1\#3) && (cast(\_1\#3 as double) > 1.0))  
　　　　　　+- SerializeFromObject \[staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input\[0, scala.Tuple2, true\]).\_1, true) AS \_1\#3, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input\[0, scala.Tuple2, true\]).\_2, true) AS \_2\#4\]  
　　　　　　　　+- ExternalRDD \[obj\#2\]

说明：和之前的一样，优化的部分就是过滤条件，先判断是否为null（hive和关系型数据库都没有），这个和RDD的不同之处是rdd是将数据全部加在进来，而sparksql如果遇到有null值的直接停止，这个是个简单的优化方案，具体其他的优化措施还是根据所写的sql语句。

--4:物理执行计划

== Physical Plan ==

\*Filter (isnotnull(\_1\#3) && (cast(\_1\#3 as double) > 1.0))  
　　　　　　+- \*SerializeFromObject \[staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input\[0, scala.Tuple2, true\]).\_1, true) AS \_1\#3, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input\[0, scala.Tuple2, true\]).\_2, true) AS \_2\#4\]  
*　　　　　　　　*\+- Scan ExternalRDDScan\[obj\#2\]

说明：同上所说的物理执行计划。