Spark——传递函数与闭包

Posted 2022-07-04 sysocjs

        在Scala中，你可以在任何作用于内定义函数，在函数体内，可以访问相应作用域内的任何变量；还不止，你的函数还可以在变量不再处于作用于内的时候被调用，这就是闭包的最基本的理解。

一、transform、action算子的函数参数

        在spark集群中，spark应用由负责运行用户编写的main函数，以及在集群上运行的各种并行操作的驱动器程序（Driver）和并行运行在集群各节点的工作进程（Executor）共同组成。action算子会触发spark提交job，在提交job的过程中，transform算子和action算子中的func会被封装成闭包，然后发送到各个worker节点上去执行（数据就近原则）。

        显然，闭包是有状态的，主要表现为那些自由变量，以及自由变量依赖到的其他变量，所以，在将一个简单的函数或者一段代码片段传递给算子作为参数前，spark会检测闭包内所有涉及的变量，然后序列化变量，再传给worker节点，再反序列化执行。（检测——序列化——传递变量——反序列化）

函数参数表示为：

val f:(Double)=>Double = 2*_

f的类型是（Double）=> Double，传入一个Double类型参数，返回一个Double类型的值。spark的transform和action算子都用到了函数参数，这其中闭包的运用最频繁。

val f(x:Int) = (x:Int) => 2*x
val rdd = sc.parallelize(1 to 10)
val rdd1 = rdd.map(x => f(x))

结果rdd1的值为Array(2,4,6,8,10,12,14,16,18,20)，这似乎没有涉及到什么闭包的知识点，不要着急，这里先介绍transform、action算子是怎样调用函数参数的。

二、闭包的理解

def mulBy (factor : Double) = (x:Double) => factor * x
val triple = mulBy(3)
val half = mulBy(0.5)
println(s"$triple(14), $half(14)")

         定义了一个函数mulBy，类型是 Double，值为(x:Double) => factor * x；

        首先，mulBy的首次被调用，将参数3传给(x:Double) => factor * x，factor=3，该变量在mulBy被引用，并将函数参数存入triple。然后参数变量factor从运行时的栈上被弹出；

        然后，mulBy再次被调用，factor的值被设置为0.5，同样的，新的参数函数存入half中，参数变量factor从运行时的栈上被弹出；

        因为每次调用mulBy函数后，都将其值存入到一个变量中（如上面的triple和half），当使用triple函数或half函数时factor相当于是作用域外，这就是“闭包”，闭包由代码和代码用到的任何非局部变量定义构成。因此，输出结果为：42，7

        虽然表象上triple和half的调用，仍然使用factor变量，但可以理解为，triple和half函数的factor并不是一个变量，而是真实的、不变的一个常量值3和0.5。

三、闭包进一步理解：spark本地模式 VS 集群模式

        通过上面可以理解spark算子怎样遍历调用一个函数，函数涉及的变量如何到达worker节点，以及闭包的概念。当一个集群上执行代码时，变量和方法的范围以及生命周期，是spark比较难理解的地方。

var counter = 0
var rdd = sc.parallelize(data)
rdd.foreach(x=>counter += x)
println(s"Counter value : $counter")

        对于单纯的RDD元素总和，根据是否运行在同一个虚拟机上，他们表现的行为完全不同。

        在本地模式下，在某些情况下，驱动程序会运行在同一个JVM内，即各个程序操作的counter属于同一个，从而可以得到“预期”的RDD元素总和结果。

        在集群模式下，为了执行作业，spark将RDD分拆成多个task，每个task由一个执行器（Executor，即一个task只能被一个Executor消化，一个Executor可以消化多个task）执行操作。在执行前，spark计算闭包（检测闭包变量和方法，上述代码指的是counter和foreach），这个闭包会被序列化，并分发给每一个执行器。换句话说，每个执行器得到各自的counter，对counter进行修改时，也只是修改自己的counter，而驱动器（Driver）上的counter并没有被修改，所以最终的counter输出结果没有达到预期，输出为0。这个可以理解为Driver的counter变量是全局变量，Executor的counter是局部变量。

        所以Spark为了应对这种由于闭包产生的影响，支持定义使用全局共享变量，广播（broadcast）变量，用来将一个值缓存到所有节点的内存中。对于累加操作，还可以使用累加器（accumulator）。

var accum = sc.accumulator(0)
val value = sc.parallelize(Array(1,2,3,4)).foreach(x => accum+=x).value
println(s"accum = $accum")
//accum的输出结果为10