spark 为什么要用broadcast[转]

为什么要用broadcast?

If you have huge array that is accessed from Spark Closures, for example some reference data, this array will be shipped to each spark node with closure. For example if you have 10 nodes cluster with 100 partitions (10 partitions per node), this Array will be distributed at least 100 times (10 times to each node).

If you use broadcast it will be distributed once per node using efficient p2p protocol.

val array: Array[Int] = ??? // some huge array
val broadcasted = sc.broadcast(array)

And some RDD

val rdd: RDD[Int] = ???

In this case array will be shipped with closure each time

rdd.map(i => array.contains(i))

and with broadcast you'll get huge performance benefit

rdd.map(i => broadcasted.value.contains(i))

变量不broadcast仅仅影响的是效率吗？

理解闭包
Spark中理解起来比较困难的一点是当代码在集群上运行时变量和方法的生命周期和作用域(scope)。当作用于RDD上的操作修改了超出它们作用域范围的变量时，会引起一些混淆。为了说明这个问题，使用下面的例子。该例中使用foreach()，对counter(计数器)进行增加，相同的问题也会发生在其他操作中。

例子

下面的例子在以本地模式运行(--master = local[n]) 和将它部署到集群中 (例如通过 spark-submit 提交到 YARN)对比发现会产生不同的结果。

1
2
3
4
5

var counter = 0
var rdd = sc.parallelize(data)
// 错误,请不要这样做！！
rdd.foreach(x => counter += x)
println("Counter value: " + counter)

本地模式 vs. 集群模式

这里主要的挑战是上面代码的行为是有歧义的。以本地模式运行在单个JVM上，上面的代码会将RDD中的值进行累加，并且将它存储到counter中。这是因为RDD和变量counter在driver节点的相同内存空间中。
然而，以集群模式运行时，会更加复杂，上面的代码的结果也许不会如我们预期的那样。当执行一个作业(job)时,Spark会将RDD分成多个任务(task)--每一个任务都会由一个executor来执行。在执行之前，Spark会计算闭包(closure)。闭包是对executors可见的那部分变量和方法，executors会用闭包来执行RDD上的计算(在这个例子中，闭包是foreach())。这个闭包是被序列化的，并且发送给每个executor。在本地模式中，只有一个executor，所以共享相同的闭包。然而，在集群模式中，就不是这样了。executors会运行在各自的worker节点中，每个executor都有闭包的一个复本。
发送给每个executor的闭包中的变量其实也是复本。每个foreach函数中引用的counter不再是driver节点上的counter。当然，在driver节点的内存中仍然存在这一个counter，但是这个counter对于executors来说是不可见的。executors只能看到自己的闭包中的复本。这样，counter最后的值仍旧是0，因为所有在counter的操作只引用了序列化闭包中的值。
为了在这样的场景中，确保这些行为正确，应该使用累加变量(Accumulator)。在集群中跨节点工作时，Spark中的累加变量提供了一种安全的机制来更新变量。所以可变的全局状态应该使用累加变量来定义。

所以上面的例子可以这样写：

1
2
3
4
5

// counter现在是累加变量
var counter = sc.accumulator(0)
var rdd = sc.parallelize(data)
rdd.foreach(x => counter += x)
println("Counter value: " + counter)