Spark一个非常重要的特性就是共享变量。

默认情况下,如果在一个算子的函数中使用到了某个外部的变量,那么这个变量的值会被拷贝到每个task中,此时每个task只能操作自己的那份变量副本。如果多个task想要共享某个变量,那么这种方式是做不到的。

Spark为此提供了两种共享变量,一种是Broadcast Variable(广播变量),另一种是Accumulator(累加变量)。Broadcast Variable会将用到的变量,仅仅为每个节点拷贝一份,即每个Executor拷贝一份,更大的用途是优化性能,减少网络传输以及内存损耗。Accumulator则可以让多个task共同操作一份变量,主要可以进行累加操作。Broadcast Variable是共享读变量,task不能去修改它,而Accumulator可以让多个task操作一个变量。

广播变量

广播变量允许编程者在每个Executor上保留外部数据的只读变量,而不是给每个任务发送一个副本。

每个task都会保存一份它所使用的外部变量的副本,当一个Executor上的多个task都使用一个大型外部变量时,对于Executor内存的消耗是非常大的,因此,我们可以将大型外部变量封装为广播变量,此时一个Executor保存一个变量副本,此Executor上的所有task共用此变量,不再是一个task单独保存一个副本,这在一定程度上降低了Spark任务的内存占用。

img

task使用外部变量

img

使用广播变量

Spark还尝试使用高效的广播算法分发广播变量,以降低通信成本。

Spark提供的Broadcast Variable是只读的,并且在每个Executor上只会有一个副本,而不会为每个task都拷贝一份副本,因此,它的最大作用,就是减少变量到各个节点的网络传输消耗,以及在各个节点上的内存消耗。此外,Spark内部也使用了高效的广播算法来减少网络消耗。

可以通过调用SparkContext的broadcast()方法来针对每个变量创建广播变量。然后在算子的函数内,使用到广播变量时,每个Executor只会拷贝一份副本了,每个task可以使用广播变量的value()方法获取值。

在任务运行时,Executor并不获取广播变量,当task执行到 使用广播变量的代码时,会向Executor的内存中请求广播变量,如下图所示:

img

task向Executor请求广播变量

之后Executor会通过BlockManager向Driver拉取广播变量,然后提供给task进行使用,如下图所示:

img

Executor从Driver拉取广播变量

广播大变量是Spark中常用的基础优化方法,通过减少内存占用实现任务执行性能的提升。

累加器

累加器(accumulator):Accumulator是仅仅被相关操作累加的变量,因此可以在并行中被有效地支持。它们可用于实现计数器(如MapReduce)或总和计数。

Accumulator是存在于Driver端的,集群上运行的task进行Accumulator的累加,随后把值发到Driver端,在Driver端汇总(Spark UI在SparkContext创建时被创建,即在Driver端被创建,因此它可以读取Accumulator的数值),由于Accumulator存在于Driver端,从节点读取不到Accumulator的数值。

Spark提供的Accumulator主要用于多个节点对一个变量进行共享性的操作。Accumulator只提供了累加的功能,但是却给我们提供了多个task对于同一个变量并行操作的功能,但是task只能对Accumulator进行累加操作,不能读取它的值,只有Driver程序可以读取Accumulator的值。

Accumulator的底层原理如下图所示:

img

累加器原理