Spark编程指南入门之Java篇七-共享变量

11. 共享变量

通常情况下，当一个传递给Spark操作的函数(例如map或者reduce)在远程集群节点执行时，函数使用的所有变量都是原变量的副本。这些变量被复制到集群的每一台服务器，在各个远程服务器更新的变量是不会更新回驱动节点程序的。不同任务读写共享变量的效率是比较低的，然而，Spark提供了2种限制类型的共享变量，广播变量和累加器。

11.1 广播变量

广播变量允许编程者将一个只读变量缓存在每一台服务器而不是给每个任务都传递一个副本。它们可以被用于，例如，以一种高效的方式给每个节点传递一份大的数据集副本。Spark也尝试使用高效的广播算法分发广播变量，以降低通讯成本。

Spark的actions是通过一系列的阶段执行的，由分布的shuffle操作分隔。Spark在每个阶段自动地广播任务需要的公共数据，这种方式广播的数据以序列化形式缓存，并在执行每个任务前进行反序列化。这明确地说明了，仅当垮阶段的任务需要相同的数据或者以反序列化形式缓存数据是重要的情况下，创建广播变量才有用。

广播变量可以通过调用SparkContext.broadcast(v)进行创建，它是一个普通变量v的封装器，它的值可以通过调用value()方法读取：

Broadcast<int[]> broadcastVar = sc.broadcast(new int[] {1, 2, 3});

broadcastVar.value();
// 返回 [1, 2, 3]

广播变量被创建后，在集群上执行的任何函数都应该使用该广播变量而不是原来的值v，以便原来的值v不会被传递到各节点多于一次。另外，为了保证所有节点读取相同的广播变量，对象v在被广播后不应该再修改。

11.2 累加器

累加器是通过关联交换操作只“加”的变量，因此可以有效地支持并行操作。它可用于实现计数器或求和，Spark原生支持数值类型的累加器，编程者可以添加对新类型的支持。在下图中，Spark的web UI会显示累加器变化的每个阶段，在“Tasks”表里会显示每一个被任务修改的累加器的值。



在UI上跟踪累加器的变化可以有助于理解其运行的原理(注意：Python暂时还不支持)

一个数值累加器可以通过调用SparkContext.longAccumulator()或者SparkContext.doubleAccumulator()方法创建，分别用于累加对应的Long或者Double类型的值。在集群上运行的任务可以使用add方法进行累加操作。但是，任务本身不能读取累加器的值，只有驱动程序才能调用value方法读取。

下面的代码使用累加器对数组的元素进行累加：

LongAccumulator accum = jsc.sc().longAccumulator();

jsc.parallelize(Arrays.asList(1, 2, 3, 4)).foreach(x -> accum.add(x));

accum.value();
// 返回 10

上述代码使用了内置支持的Long类型累加器，编程者也可以通过实现AccumulatorV2创建自定义的累加器类型。AccumulatorV2这个抽象类有几个必须重写的方法：reset方法重置累加器为0、add方法求和、merge方法合并2个累加器等，其余必须重写的方法请查看API文档。例如，如果有一个表示数学矢量的类MyVector，我们可以开发如下代码：

class VectorAccumulatorV2 extends AccumulatorV2<MyVector, MyVector> {

private MyVector myVector = MyVector.createZeroVector();

public void reset() {
myVector.reset();
}

public void add(MyVector v) {
myVector.add(v);
}
...
}

// 创建该类型的累加器:
VectorAccumulatorV2 myVectorAcc = new VectorAccumulatorV2();
// 在spark context注册:
jsc.sc().register(myVectorAcc, "MyVectorAcc1");

注意，当我们创建自定义的累加器类型，其运算结果的类型可以与累加的元素类型不同。

对于只在actions操作中更新的累加器，Spark保证每一个任务只更新累加器一次，例如，重新启动任务是不会再更新累加器的值。在转换操作中，应该注意的是如果job的阶段被重新执行的话，每一个任务的更新会执行不止一次。

累加器不会改变Spark的延迟加载的模式。如果累加器在RDD的一个操作中进行更新，其值只会在RDD执行action操作的时候更新。因此，当在一个像map()的转换操作中，累加器的更新是不保证被执行的。下面的代码展示了这样的特性：

LongAccumulator accum = jsc.sc().longAccumulator();
data.map(x -> { accum.add(x); return f(x); });
// 这里，accum仍然是0，因为没有执行action操作

END O(∩_∩)O