scala – 了解Spark中的闭包和并行性

我试着理解Spark中某些东西是如何工作的.在示例中,如 http://spark.apache.org/docs/latest/programming-guide.html#understanding-closures-a-nameclosureslinka所示

说代码将对RDD中的值求和并将其存储在计数器中,但这不是这种情况,因为它不起作用.只有删除了paralelize,才有效.

有人可以向我解释这是如何工作的吗？或者示例错了？

谢谢

val data = Array(1,2,3,4,5)
var counter = 0
var rdd = sc.parallelize(data)

// Wrong: Don't do this!!
rdd.foreach(x => counter += x)

println("Counter value: " + counter)

解决方法

尼克的例子和上面提供的解释绝对正确,让我深入解释一下 – >

让我们假设我们正在使用单个工作节点和执行器来处理单个节点,并且我们使用foreach而不是RDD来计算RDD中的元素数量.我们知道我们在一个节点上,因此数据不会被分发并且将保持单一身份,因此计数变量(Closure – >这类变量称为Closure)将计入每个元素,并且此更新将每当发生增量时,都会被发送到执行程序,然后执行程序将关闭提交给驱动程序节点.

Drivernode – >执行程序和驱动程序都将驻留在单个节点上,因此驱动程序节点的计数变量将位于执行程序节点的范围内,因此将更新驱动程序节点计数变量值.

我们已经从驱动程序节点提供了结果计数值,而不是从执行程序节点提供.

Executor -> closure -> data

现在假设我们在集群环境中工作,假设有2个节点和2个工作者和执行者.现在数据将被分成几个部分,因此 – >

Data -> Data_1,Data_2

Drivernode – >在不同的节点上有它的计数变量但对Executor 1和Executor 2不可见,因为它们驻留在不同的节点上,因此executor1和executor2不能更新驱动程序节点上的count变量

Executor1-> processing(Data_1) with closure_1
Executor2-> processing(Data_1) with closure_2

闭包1将更新执行器1,因为它可序列化为执行器1,类似地,闭包2将更新执行器2

为了解决这种情况,我们使用Accumulator,如下所示：

val counter=sc.accumulator(0)