通常当一个函数传入Spark的操作（比如 map or reduce）会在集群的一个节点上执行。函数是在相互分开的变量的副本上运行。这些变量被复制到每个机器，并且远程机器上的更新不会传回dirver的进程。通常来讲，在不同的task之间读写共享变量非常低效。但是Spark提供了两个限制性的共享变量：广播变量和累加器（broadcast variables and accumulators.）。
广播变量使用（ Broadcast Variables）
广播变量可以允许开发者在每台机器上保存一个只读的变量，而不是在task之间传输该变量的副本。广播变量可以用来以比较高效的方式给每个节点一个比较大的数据集的副本。Spark还使用了高效的广播算法分发广播变量来减少通讯的成本。
Spark的action是执行在stages上并且被shuffle切分。Spark会自动的广播普通的每个stage的task所需要的数据。这种方式广播的数据是以序列化的方式进行缓存的，并且在每个task运行的时候进行反序列化。也就说，只有当多个stage的task需要相同数据或者缓存的数据是十分重要的时候，创建广播变量才是有用的。这儿我补充一个点，官方文档只是简单说了下，广播变量可以比较搞笑的方式给每个节点提供一个比较大的数据。但是没有说具体的实现。
具体的实现方式是这样的：
driver会把序列化后的对象切分成小的chunk（块），并且把这些chunk保存在driver的BlockManager中。每个Executor 上，executor 第一次会尝试从它的BlockManager上读取这个对象。如果不存在，它就会从远程的driver或者其他executors上（如果其他executor上可以的话）。一旦获取到chunks则会将chunks保存在自己BlockManager中，其他executor就可以从本机获取了。
这样避免了在往每一个executor上广播数据的时候driver成为瓶颈。
通过调用SparkContext.broadcast(v) 可以创建一个v的广播变量。这个广播变量是一个v的包装类，它的值可以通过value方法进行访问。代码如下：
scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)
scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)
广播变量创建后，运行在集群上的任何方法必须使用创建的广播变量来代替v这个值。这样v就不会多次被在节点间进行移动。另外，为了确保每个节点获取到相同的广播变量的值，对象v在广播之后不应该被改变。
累加器使用
累加器是通过交替的操作可以增加的变量，并且可以运行在并行的情况下。可以用来实现一个计数器（和 MapReduce中的一样）或者求和。Spark天然支持数字类型的累加器，开发人员可以添加新类型的支持。
用户可以创建一个有名字或者无名字的累加器。如下图所示，一个有名字的累加器（实例中是 counter）会在Stage 的web ui 界面看到。Spark会在累加器被每个task修改的时候展示这个值。
image.png
跟踪UI中的累加器对于理解正在运行的stage进程会很有帮助。（Python不持之目前）
累加器使用实例
集群中运行的task可以调用add方法 进行累加，但是不能读取。只有Diver进程可以通过value方法读取累加器的值。
以下代码演示了一个数组的累加
scala> val accum = sc.longAccumulator("My Accumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s
scala> accum.value
res2: Long = 10
自定义累加器
上面的代码使用内在支持的Long类型的累加器。开发者可以通过继承AccumulatorV2.创建自己的类型。AccumulatorV2 抽象类有一些方法，其中必须被重写的有：reset 设置累加器的值未0.add将一个值加到累加器中，merge 将一个相同类型的累加器合并成一个。其他必须被重写的方法在API documentation.举个例子，如果我们有一个MyVector代表数学向量，可以这么写：
class VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] {
private val myVector: MyVector = MyVector.createZeroVector
def reset(): Unit = {
myVector.reset()
}
def add(v: MyVector): Unit = {
myVector.add(v)
}
...
}
// Then, create an Accumulator of this type:
val myVectorAcc = new VectorAccumulatorV2
// Then, register it into spark context:
sc.register(myVectorAcc, "MyVectorAcc1")
需要注意一点，当开发者自定义了累加器的时候，累加的结果是可以和累加的元素类型不同的。
累加器的更新只会在action的操作当中，Spark会保证每个task对累加器的更新只会执行一次，举个例子，重新执行的task不会更新累加器的值。在transform的操作中，开发者需要主要的是，如果task或者stage重新执行，每个task的对累加器的更新都会执行多次。
在Spark的延迟计算模式下累加器的值不会改变。如果累加器在一个RDD上的操作中被改变了，他的值只会被改变一次。也就是说，当在延迟计算的transformation 比如map中，累加器不会保证被执行。以下的代码能够演示这个特征：
val accum = sc.longAccumulator
data.map { x => accum.add(x); x }
// Here, accum is still 0 because no actions have caused the map operation to be computed.
网友评论