AI 助理
备案控制台
开发者社区 大数据 文章 正文

Spark修炼之道（进阶篇）——Spark入门到精通：第九节 Spark SQL运行流程解析

2015-12-31 3491
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
公共DNS（含HTTPDNS解析），每月1000万次HTTP解析
全局流量管理 GTM，标准版 1个月
云解析 DNS，旗舰版 1个月
简介： 1.整体运行流程使用下列代码对SparkSQL流程进行分析，让大家明白LogicalPlan的几种状态，理解SparkSQL整体执行流程// sc is an existing SparkContext.val sqlContext = new org.apache.spark.sql.SQLContext(sc)// this is used to implic

1.整体运行流程

使用下列代码对SparkSQL流程进行分析，让大家明白LogicalPlan的几种状态，理解SparkSQL整体执行流程

// 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")

（1）查看teenagers的Schema信息

scala> teenagers.printSchema
root
 |-- name: string (nullable = true)
 |-- age: integer (nullable = false)

（2）查看运行流程

scala> teenagers.queryExecution
res3: org.apache.spark.sql.SQLContext#QueryExecution =
== Parsed Logical Plan ==
'Project [unresolvedalias('name),unresolvedalias('age)]
 'Filter (('age >= 13) && ('age <= 19))
  'UnresolvedRelation [people], None

== Analyzed Logical Plan ==
name: string, age: int
Project [name#0,age#1]
 Filter ((age#1 >= 13) && (age#1 <= 19))
  Subquery people
   LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== Optimized Logical Plan ==
Filter ((age#1 >= 13) && (age#1 <= 19))
 LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== Physical Plan ==
Filter ((age#1 >= 13) && (age#1 <= 19))
 Scan PhysicalRDD[name#0,age#1]

Code Generation: true

QueryExecution中表示的是整体Spark SQL运行流程，从上面的输出结果可以看到，一个SQL语句要执行需要经过下列步骤：

== （1）Parsed Logical Plan ==
'Project [unresolvedalias('name),unresolvedalias('age)]
 'Filter (('age >= 13) && ('age <= 19))
  'UnresolvedRelation [people], None

== （2）Analyzed Logical Plan ==
name: string, age: int
Project [name#0,age#1]
 Filter ((age#1 >= 13) && (age#1 <= 19))
  Subquery people
   LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== （3）Optimized Logical Plan ==
Filter ((age#1 >= 13) && (age#1 <= 19))
 LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== （4）Physical Plan ==
Filter ((age#1 >= 13) && (age#1 <= 19))
 Scan PhysicalRDD[name#0,age#1]

//启动动态字节码生成技术（bytecode generation，CG），提升查询效率
Code Generation: true

2.全表查询运行流程

执行语句：

val all= sqlContext.sql("SELECT * FROM people")

运行流程：

scala> all.queryExecution
res9: org.apache.spark.sql.SQLContext#QueryExecution =
//注意*号被解析为unresolvedalias(*)
== Parsed Logical Plan ==
'Project [unresolvedalias(*)]
 'UnresolvedRelation [people], None

== Analyzed Logical Plan ==
//unresolvedalias(*)被analyzed为Schema中所有的字段
//UnresolvedRelation [people]被analyzed为Subquery people
name: string, age: int
Project [name#0,age#1]
 Subquery people
  LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== Optimized Logical Plan ==
LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== Physical Plan ==
Scan PhysicalRDD[name#0,age#1]

Code Generation: true

3. filter查询运行流程

执行语句：

scala> val filterQuery= sqlContext.sql("SELECT * FROM people WHERE age >= 13 AND age <= 19")
filterQuery: org.apache.spark.sql.DataFrame = [name: string, age: int]

执行流程：

scala> filterQuery.queryExecution
res0: org.apache.spark.sql.SQLContext#QueryExecution =
== Parsed Logical Plan ==
'Project [unresolvedalias(*)]
 'Filter (('age >= 13) && ('age <= 19))
  'UnresolvedRelation [people], None

== Analyzed Logical Plan ==
name: string, age: int
Project [name#0,age#1]
 //多出了Filter，后同
 Filter ((age#1 >= 13) && (age#1 <= 19))
  Subquery people
   LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:20

== Optimized Logical Plan ==
Filter ((age#1 >= 13) && (age#1 <= 19))
 LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:20

== Physical Plan ==
Filter ((age#1 >= 13) && (age#1 <= 19))
 Scan PhysicalRDD[name#0,age#1]

Code Generation: true

4. join查询运行流程

执行语句：

val joinQuery= sqlContext.sql("SELECT * FROM people a, people b where a.age=b.age")

查看整体执行流程

scala> joinQuery.queryExecution
res0: org.apache.spark.sql.SQLContext#QueryExecution =
//注意Filter
//Join Inner
== Parsed Logical Plan ==
'Project [unresolvedalias(*)]
 'Filter ('a.age = 'b.age)
  'Join Inner, None
   'UnresolvedRelation [people], Some(a)
   'UnresolvedRelation [people], Some(b)

== Analyzed Logical Plan ==
name: string, age: int, name: string, age: int
Project [name#0,age#1,name#2,age#3]
 Filter (age#1 = age#3)
  Join Inner, None
   Subquery a
    Subquery people
     LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22
   Subquery b
    Subquery people
     LogicalRDD [name#2,age#3], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== Optimized Logical Plan ==
Project [name#0,age#1,name#2,age#3]
 Join Inner, Some((age#1 = age#3))
  LogicalRDD [name#0,age#1], MapPartitionsRDD[4]...

//查看其Physical Plan
scala> joinQuery.queryExecution.sparkPlan
res16: org.apache.spark.sql.execution.SparkPlan =
TungstenProject [name#0,age#1,name#2,age#3]
 SortMergeJoin [age#1], [age#3]
  Scan PhysicalRDD[name#0,age#1]
  Scan PhysicalRDD[name#2,age#3]

前面的例子与下面的例子等同，只不过其运行方式略有不同，执行语句：

scala> val innerQuery= sqlContext.sql("SELECT * FROM people a inner join people b on a.age=b.age")
innerQuery: org.apache.spark.sql.DataFrame = [name: string, age: int, name: string, age: int]

查看整体执行流程：

scala> innerQuery.queryExecution
res2: org.apache.spark.sql.SQLContext#QueryExecution =
//注意Join Inner
//另外这里面没有Filter
== Parsed Logical Plan ==
'Project [unresolvedalias(*)]
 'Join Inner, Some(('a.age = 'b.age))
  'UnresolvedRelation [people], Some(a)
  'UnresolvedRelation [people], Some(b)

== Analyzed Logical Plan ==
name: string, age: int, name: string, age: int
Project [name#0,age#1,name#4,age#5]
 Join Inner, Some((age#1 = age#5))
  Subquery a
   Subquery people
    LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22
  Subquery b
   Subquery people
    LogicalRDD [name#4,age#5], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

//注意Optimized Logical Plan与Analyzed Logical Plan
//并没有进行特别的优化，突出这一点是为了比较后面的子查询
//其Analyzed和Optimized间的区别
== Optimized Logical Plan ==
Project [name#0,age#1,name#4,age#5]
 Join Inner, Some((age#1 = age#5))
  LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder ...

//查看其Physical Plan
scala> innerQuery.queryExecution.sparkPlan
res14: org.apache.spark.sql.execution.SparkPlan =
TungstenProject [name#0,age#1,name#6,age#7]
 SortMergeJoin [age#1], [age#7]
  Scan PhysicalRDD[name#0,age#1]
  Scan PhysicalRDD[name#6,age#7]

5. 子查询运行流程

执行语句：

scala> val subQuery=sqlContext.sql("SELECT * FROM (SELECT * FROM people WHERE age >= 13)a where a.age <= 19")
subQuery: org.apache.spark.sql.DataFrame = [name: string, age: int]

查看整体执行流程：


scala> subQuery.queryExecution
res4: org.apache.spark.sql.SQLContext#QueryExecution =
== Parsed Logical Plan ==
'Project [unresolvedalias(*)]
 'Filter ('a.age <= 19)
  'Subquery a
   'Project [unresolvedalias(*)]
    'Filter ('age >= 13)
     'UnresolvedRelation [people], None

== Analyzed Logical Plan ==
name: string, age: int
Project [name#0,age#1]
 Filter (age#1 <= 19)
  Subquery a
   Project [name#0,age#1]
    Filter (age#1 >= 13)
     Subquery people
      LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

//这里需要注意Optimized与Analyzed间的区别
//Filter被进行了优化
== Optimized Logical Plan ==
Filter ((age#1 >= 13) && (age#1 <= 19))
 LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== Physical Plan ==
Filter ((age#1 >= 13) && (age#1 <= 19))
 Scan PhysicalRDD[name#0,age#1]

Code Generation: true

6. 聚合SQL运行流程

执行语句：

scala> val aggregateQuery=sqlContext.sql("SELECT a.name,sum(a.age) FROM (SELECT * FROM people WHERE age >= 13)a where a.age <= 19 group by a.name")
aggregateQuery: org.apache.spark.sql.DataFrame = [name: string, _c1: bigint]

运行流程查看：


scala> aggregateQuery.queryExecution
res6: org.apache.spark.sql.SQLContext#QueryExecution =
//注意'Aggregate ['a.name], [unresolvedalias('a.name),unresolvedalias('sum('a.age))]
//即group by a.name被 parsed为unresolvedalias('a.name)
== Parsed Logical Plan ==
'Aggregate ['a.name], [unresolvedalias('a.name),unresolvedalias('sum('a.age))]
 'Filter ('a.age <= 19)
  'Subquery a
   'Project [unresolvedalias(*)]
    'Filter ('age >= 13)
     'UnresolvedRelation [people], None

== Analyzed Logical Plan ==
name: string, _c1: bigint
Aggregate [name#0], [name#0,sum(cast(age#1 as bigint)) AS _c1#9L]
 Filter (age#1 <= 19)
  Subquery a
   Project [name#0,age#1]
    Filter (age#1 >= 13)
     Subquery people
      LogicalRDD [name#0,age#1], MapPartitionsRDD[4] at rddToDataFrameHolder at <console>:22

== Optimized Logical Plan ==
Aggregate [name#0], [name#0,sum(cast(age#1 as bigint)) AS _c1#9L]
 Filter ((age#1 >= 13) && (age#1 <= 19))
  LogicalRDD [name#0,age#1], MapPartitions...

//查看其Physical Plan
scala> aggregateQuery.queryExecution.sparkPlan
res10: org.apache.spark.sql.execution.SparkPlan =
TungstenAggregate(key=[name#0], functions=[(sum(cast(age#1 as bigint)),mode=Final,isDistinct=false)], output=[name#0,_c1#14L])
 TungstenAggregate(key=[name#0], functions=[(sum(cast(age#1 as bigint)),mode=Partial,isDistinct=false)], output=[name#0,currentSum#17L])
  Filter ((age#1 >= 13) && (age#1 <= 19))
   Scan PhysicalRDD[name#0,age#1]

其它SQL语句，大家可以使用同样的方法查看其执行流程，以掌握Spark SQL背后实现的基本思想。

文章标签：
云解析DNS
分布式计算
Spark
SQL
Scala
关键词：
云解析DNS流程
云解析DNS sql
SQL解析
sql云解析DNS
apache spark SQL
周志湖
目录
相关文章
bruce_xiaowei
|
9天前
|
数据挖掘 vr&ar C++
让UE自动运行Python脚本：实现与实例解析
本文介绍如何配置Unreal Engine（UE）以自动运行Python脚本，提高开发效率。通过安装Python、配置UE环境及使用第三方插件，实现Python与UE的集成。结合蓝图和C++示例，展示自动化任务处理、关卡生成及数据分析等应用场景。
bruce_xiaowei
61 5
潇洒洒
|
1月前
|
SQL IDE 数据库连接
IntelliJ IDEA处理大文件SQL：性能优势解析
在数据库开发和管理工作中，执行大型SQL文件是一个常见的任务。传统的数据库管理工具如Navicat在处理大型SQL文件时可能会遇到性能瓶颈。而IntelliJ IDEA，作为一个强大的集成开发环境，提供了一些高级功能，使其在执行大文件SQL时表现出色。本文将探讨IntelliJ IDEA在处理大文件SQL时的性能优势，并与Navicat进行比较。
潇洒洒
34 4
游客ny3yjjsyp7iro
|
1月前
|
SQL Java 数据库连接
canal-starter 监听解析 storeValue 不一样，同样的sql 一个在mybatis执行 一个在数据库操作，导致解析不出正确对象
canal-starter 监听解析 storeValue 不一样，同样的sql 一个在mybatis执行 一个在数据库操作，导致解析不出正确对象
游客ny3yjjsyp7iro
46 0
1100237741946300
|
2月前
|
SQL 监控 数据库
SQL语句是否都需要解析及其相关技巧和方法
在数据库管理中，SQL（结构化查询语言）语句的使用无处不在，它们负责数据的查询、插入、更新和删除等操作
1100237741946300
38 1
赵渝强老师
|
1月前
|
SQL JSON 分布式计算
【赵渝强老师】Spark SQL的数据模型：DataFrame
本文介绍了在Spark SQL中创建DataFrame的三种方法。首先，通过定义case class来创建表结构，然后将CSV文件读入RDD并关联Schema生成DataFrame。其次，使用StructType定义表结构，同样将CSV文件读入RDD并转换为Row对象后创建DataFrame。最后，直接加载带有格式的数据文件（如JSON），通过读取文件内容直接创建DataFrame。每种方法都包含详细的代码示例和解释。
赵渝强老师
49 0
陌陌谣
|
1月前
|
SQL 监控 安全
员工上网行为监控软件：SQL 在数据查询监控中的应用解析
在数字化办公环境中，员工上网行为监控软件对企业网络安全和管理至关重要。通过 SQL 查询和分析数据库中的数据，企业可以精准了解员工的上网行为，包括基础查询、复杂条件查询、数据统计与分析等，从而提高网络管理和安全防护的效率。
陌陌谣
30 0
游客g6r3tt6lma5xg
|
2月前
|
SQL 数据库
SQL数据库基础语法入门
[link](http://www.vvo.net.cn/post/082935.html)
游客g6r3tt6lma5xg
46 0
1100237741946300
|
2月前
|
SQL 数据可视化 BI
SQL语句及查询结果解析：技巧与方法
在数据库管理和数据分析中，SQL语句扮演着至关重要的角色
1100237741946300
458 0
1100237741946300
|
2月前
|
SQL 监控 关系型数据库
SQL错误代码1303解析与处理方法
在SQL编程和数据库管理中，遇到错误代码是常有的事，其中错误代码1303在不同数据库系统中可能代表不同的含义
1100237741946300
123 0
1100237741946300
|
2月前
|
SQL 存储 关系型数据库
SQL默认索引是什么：深入解析与技巧
在SQL数据库中，索引是一种用于提高查询性能的重要数据结构
1100237741946300
54 0

热门文章

最新文章

  • 1
    Flask 框架防止 SQL 注入攻击的方法
  • 2
    Vanna：开源 AI 检索生成框架，自动生成精确的 SQL 查询
  • 3
    SQL自学笔记（3）：SQL里的DCL,DQL都代表什么？
  • 4
    Flink SQL Deduplication 去重以及如何获取最新状态操作
  • 5
    MySQL导入.sql文件后数据库乱码问题
  • 6
    MySQL进阶突击系列(02)一条更新SQL执行过程 | 讲透undoLog、redoLog、binLog日志三宝
  • 7
    MySQL 高级(进阶) SQL 语句
  • 8
    MySQL进阶突击系列(01)一条简单SQL搞懂MySQL架构原理 | 含实用命令参数集
  • 9
    SQL自学笔记（1）：什么是SQL？有什么用？
  • 10
    SQL慢查询优化策略
  • 1
    spark用于分析数据并将数据保存到数据库中是
    78
  • 2
    数据分享|Python、Spark SQL、MapReduce决策树、回归对车祸发生率影响因素可视化分析
    100
  • 3
    如何解决Spark在深度学习中的资源消耗问题？
    96
  • 4
    Spark在深度学习中的优缺点是什么？
    271
  • 5
    使用Spark进行机器学习
    92
  • 6
    Spark是一个基于内存的通用数据处理引擎，可以进行大规模数据处理和分析
    150
  • 7
    【Spark】Spark基础教程知识点
    96
  • 8
    DataWorks产品使用合集之在DataWorks中，通过spark访问外网的步骤如何解决
    189
  • 9
    MaxCompute操作报错合集之大数据计算的MaxCompute Spark引擎无法读取到表，是什么原因
    155
  • 10
    MaxCompute操作报错合集之spark3.1.1通过resource目录下的conf文件配置，报错如何解决
    89

    • 相关课程

    更多
  • 大数据实战项目：反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第一阶段
  • 大数据实战项目：反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第二阶段
  • 深入解析Docker容器化技术
  • Java面试疑难点解析 - 面试技巧及语言基础
  • Java面试疑难点解析 - Java Web开发
  • Java面试疑难点解析 - 系统架构及项目设计

    • 相关电子书

    更多
  • 神龙云服务器产品及技术深度解析
  • 弹性创造价值：基于ECS的最佳性价比实践解析
  • 又快又稳：阿里云下一代虚拟交换机解析

    • 相关实验场景

    更多
  • PolarDB for AI：在数据库中通过SQL实现AI能力
  • 玩转MaxCompute SQL! 30分钟搞定数据分析挖掘
  • 通过云拨测对指定服务器进行Ping/DNS监测
  • 使用SQL语句实现数据插入、修改和删除操作
  • 使用SQL语句实现数据查询操作
  • 使用SQL语句管理索引

    • 推荐镜像

    更多
  • DNS
  • NTP
  • puppet
    • 下一篇
    手把手教你白嫖阿里云服务器(免费领服务器)