Spark闭包 | driver & executor程序代码执行

Spark中的闭包

闭包的作用可以理解为:函数可以访问函数外部定义的变量,但是函数内部对该变量进行的修改,在函数外是不可见的,即对函数外源变量不会产生影响。

其实,在学习Spark时,一个比较难理解的点就是,在集群模式下,定义的变量和方法作用域的范围和生命周期。这在你操作RDD时,比如调用一些函数map、foreach时,访问其外部变量进行操作时,很容易产生疑惑。为什么我本地程序运行良好且结果正确,放到集群上却得不到想要的结果呢?

首先通过下边对RDD中的元素进行求和的示例,来看相同的代码本地模式和集群模式运行结果的区别:

Spark为了执行任务,会将RDD的操作分解为多个task,并且这些task是由executor执行的。在执行之前,Spark会计算task的闭包即定义的一些变量和方法,比如例子中的counter变量和foreach方法,并且闭包必须对executor而言是可见的,这些闭包会被序列化发送到每个executor。

在集群模式下,driver和executor运行在不同的JVM进程中,发送给每个executor的闭包中的变量是driver端变量的副本。因此,当foreach函数内引用counter时,其实处理的只是driver端变量的副本,与driver端本身的counter无关。driver节点的内存中仍有一个计数器,但该变量对executor是不可见的!executor只能看到序列化闭包的副本。因此,上述例子输出的counter最终值仍然为零,因为counter上的所有操作都只是引用了序列化闭包内的值。

在本地模式下,往往driver和executor运行在同一JVM进程中。那么这些闭包将会被共享,executor操作的counter和driver持有的counter是同一个,那么counter在处理后最终值为6。

但是在生产中,我们的任务都是在集群模式下运行,如何能满足这种业务场景呢?

这就必须引出一个后续要重点讲解的概念:Accumulator即累加器。Spark中的累加器专门用于提供一种机制,用于在集群中的各个worker节点之间执行时安全地更新变量。

一般来说,closures - constructs比如循环或本地定义的方法,就不应该被用来改变一些全局状态,Spark并没有定义或保证对从闭包外引用的对象进行更新的行为。如果你这样操作只会导致一些代码在本地模式下能够达到预期的效果,但是在分布式环境下却事与愿违。如果需要某些全局聚合,请改用累加器。对于其他的业务场景,我们适时考虑引入外部存储系统、广播变量等。

 

闭包函数从产生到在executor执行经历了什么?

首先,对RDD相关的操作需要传入闭包函数,如果这个函数需要访问外部定义的变量,就需要满足一定条件(比如必须可被序列化),否则会抛出运行时异常。闭包函数在最终传入到executor执行,需要经历以下步骤:

1. driver通过反射,运行时找到闭包访问的变量,并封装成一个对象,然后序列化该对象

2. 将序列化后的对象通过网络传输到worker节点

3. worker节点反序列化闭包对象

4. worker节点的executor执行闭包函数

简而言之,就是要通过网络传递函数、然后执行,期间会经历序列化和反序列化,所以要求被传递的变量必须可以被序列化和反序列化,否则会抛类似Error:Task not serializable: java.io.NotSerializableException when calling function outside closure only on classes not objects这样的异常。即使是本地执行时,也会按照上述的步骤执行,这也是为什么不允许在RDD内部直接操作RDD的原因(SparkContext不支持序列化)。同时,在这些算子闭包内修改外部定义的变量不会被反馈到driver端。

 

driver & executor

driver是运行用户编写Application 的main()函数的地方,具体负责DAG的构建、任务的划分、task的生成与调度等。job,stage,task生成都离不开rdd自身,rdd的相关的操作不能缺少driver端的sparksession/sparkcontext。

executor是真正执行task地方,而task执行离不开具体的数据,这些task运行的结果可以是shuffle中间结果,也可以持久化到外部存储系统。一般都是将结果、状态等汇集到driver。但是,目前executor之间不能互相通信,只能借助第三方来实现数据的共享或者通信。

 

编写的Spark程序代码,运行在driver端还是executor端呢?

先看个简单例子:通常我们在本地测试程序的时候,要打印RDD中的数据。

在本地模式下,直接使用rdd.foreach(println)或rdd.map(println)在单台机器上,能够按照预期打印并输出所有RDD的元素。

但是,在集群模式下,由executor执行输出写入的是executor的stdout,而不是driver上的stdout,所以driver的stdout不会显示这些!

要想在driver端打印所有元素,可以使用collect()方法先将RDD数据带到driver节点,然后在调用foreach(println)(但需要注意一点,由于会把RDD中所有元素都加载到driver端,可能引起driver端内存不足导致OOM。如果你只是想获取RDD中的部分元素,可以考虑使用take或者top方法)

总之,在这里RDD中的元素即为具体的数据,对这些数据的操作都是由负责task执行的executor处理的,所以想在driver端输出这些数据就必须先将数据加载到driver端进行处理。

最后做个总结:所有对RDD具体数据的操作都是在executor上执行的,所有对rdd自身的操作都是在driver上执行的。比如foreach、foreachPartition都是针对rdd内部数据进行处理的,所以我们传递给这些算子的函数都是执行于executor端的。但是像foreachRDD、transform则是对RDD本身进行一列操作,所以它的参数函数是执行在driver端的,那么它内部是可以使用外部变量,比如在SparkStreaming程序中操作offset、动态更新广播变量等。

关注微信公众号:大数据学习与分享,获取更对技术干货

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


1.SparkStreaming是什么?SparkStreaming是SparkCore的扩展API用来支持高吞吐、高容错的处理流式数据数据源可以是:Kafka、TCPsockets、Flume、Twitter等流式数据源处理数据:可以用SparkCore的算子map、reduce、join、window
本篇内容介绍了“Spark通讯录相似度计算怎么实现”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这...
本篇文章给大家分享的是有关如何进行Spark数据分析,小编觉得挺实用的,因此分享给大家学习,希望大家阅读完这篇文章后可以有所收获,话不多说...
本篇内容主要讲解“Spark Shuffle和Hadoop Shuffle有哪些区别”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“S...
这篇文章主要介绍“TSDB的数据怎么利用Hadoop/spark集群做数据分析”,在日常操作中,相信很多人在TSDB的数据怎么利用Hadoop/spark集群做数据分析问题上存在疑惑...
本篇内容介绍了“Hadoop与Spark性能原理是什么”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这
小编给大家分享一下Hadoop和Spark有什么不同,相信大部分人都还不怎么了解,因此分享这篇文章给大家参考一下,希望大家阅读完这篇文章后大有收获,下面让我们
这篇文章主要讲解了“Hadoop和Spark的Shuffle过程有什么不同”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习...
本篇文章给大家分享的是有关基于CDP7.1.1的Spark3.0技术预览版本分析是怎样的,小编觉得挺实用的,因此分享给大家学习,希望大家阅读完这篇文章后可以有所收获...
这篇文章主要介绍“Spark中foreachRDD、foreachPartition和foreach的区别是什么”,在日常操作中,相信很多人在Spark中foreachRDD、foreachPartition和foreach的...
本篇内容主要讲解“spark的动态分区裁剪怎么实现”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“spark的动态分...
本篇内容介绍了“spark的动态分区裁剪下物理计划怎么实现”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下
这篇文章给大家介绍基于Spark和TensorFlow 的机器学习实践是怎么样的,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。EMR E-Learning平台...
这篇文章将为大家详细讲解有关如何进行EMR Spark-SQL性能极致优化的分析,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识...
如何进行SparkSQL与Hive metastore Parquet转换的分析,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决...
如何浅析Hive和Spark SQL读文件时的输入任务划分,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个...
这篇文章将为大家详细讲解有关Hive on Spark参数如何调优,小编觉得挺实用的,因此分享给大家做个参考,希望大家阅读完这篇文章后可以有所收获。前言Hive on Spa...
这篇文章将为大家详细讲解有关fs.defaultFS变更使spark-sql查询hive失败是怎么回事,小编觉得挺实用的,因此分享给大家做个参考,希望大家阅读完这篇文章后可以...
这篇文章将为大家详细讲解有关怎么解析SparkCore和SparkSQL,文章内容质量较高,因此小编分享给大家做个参考,希望大家阅读完这篇文章后对相关知识有一定的了解...
怎么快速搭建Spark开发环境,针对这个问题,这篇文章详细介绍了相对应的分析和解答,希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。一,搭建本...