我试着理解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)
