Spark RDD高级应用（传参、闭包、共享变量）

不念不忘少年蓝@ 2021-09-24 12:28 235阅读 0赞

### 教程目录 ###

*  0x00 教程内容
 *  0x01 向Spark传递参数的方式
 *   *   *   *   *  1. 匿名函数
                 *  2. 传入静态方法和传入方法的引用
 *  0x02 闭包
 *   *   *   *   *  1. 闭包的概念
                 *  2. 闭包实操
                 *  3. 打印 RDD 的元素
 *  0x03 共享变量
 *   *   *   *   *  1. 广播变量
                 *  2. 累加器
 *  0xFF 总结

# 0x00 教程内容 #

1.  向Spark传递参数的两种方式
2.  闭包的概念及实操
3.  共享变量的两种方式

# 0x01 向Spark传递参数的方式 #

Spark 中的大部分操作都依赖于用户传递的函数，主要有两种方式：

方式一：匿名函数  
方式二：传入静态方法和传入方法的引用

##### 1. 匿名函数 #####

在前面的教程中，我们用到了很多次这种方式，如这句：

val wordRDD = textFileRDD.flatMap(line => line.split(" "))

`=>` 左边是`参数`，参数可以省略参数类型，右边是`函数体`。

关于匿名函数：

var inc = (x:Int) => x+1

也可以这样写，可能会更易懂一点

var inc = ((x:Int) => x+1)

或者：

var inc = { (x:Int) => x+1}

其实这句匿名函数，跟下面这种写法是一样的意思：

def add2 = new Function1[Int,Int]{   
        def apply(x:Int):Int = x+1;  
    }

使用的时候，直接把 inc 当成函数名就可以了，Java 是一门面向对象的编程语言，一切皆对象；而 Scala 是一门面向函数的编程语言，一切皆函数！

使用方式：

inc(1)

或者：

var x = inc(7)-1

我们也可以在匿名函数中可以定义多个参数：

var mul = (x: Int, y: Int) => x*y

使用方式如下：

println(mul(3, 4))

我们也可以不给匿名函数设置参数：

var userDir = () => {  System.getProperty("user.dir") }

使用方式如下：

println( userDir() )

##### 2. 传入静态方法和传入方法的引用 #####

我们可以传入`全局单例对象的静态方法`。比如：先定义 `Object Functions`，然后传递 `Functions.func1`。

val nums = sc.parallelize(List(1,2,3,4))
    object Functions{ 
        def func1(x:Int):Int = { x+1}
    }
    nums.map(Functions.func1).collect()

注意：虽然可以在**同一个类实例**中传递`方法的引用`，但是需要发送包含该类的`对象`连同`该方法`到**集群**，此过程会比仅传递`静态方法`到集群的开销更大。

# 0x02 闭包 #

##### 1. 闭包的概念 #####

`闭包`是一个函数，`返回值`依赖于声明在`函数外部`的一个或多个变量。

`闭包`可以简单的理解为：可以访问一个函数里面`局部变量`的另外一个函数。

##### 2. 闭包实操 #####

我们现在有个需求：计算所有元素的总和

val data = Array(1,2,3,4,5)
    var counter = 0
    var rdd = sc.parallelize(data)
    rdd.foreach(x => counter+=x)
    println("counter的值：" + counter)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70]  
此时发现了一个问题，计算结果竟然是0！原因是在执行的时候，Spark会将 RDD 拆分成多个 `task`，并且分发给不同的执行器（`executor`）执行，而分发到每个执行器的`counter`是复制过去的，执行器之间不能相互访问，导致执行器在执行`foreach`方法的时候只能修改自己执行器中的值，驱动程序中的`counter`值没有被修改，所以最终输出的`counter`值依然是 0。这种场景有点像函数中参数的调用，调用的时候，应该传入变量的地址，否则，当需要调用的函数执行完结果后，不会把执行后的结果返回给主函数。

在上面的代码中，我们的`foreach()` 函数就是 **闭包** 。在执行的过程中，引用了`函数外部`的`counter`变量。

那我们可以怎么去解决以上的问题呢？

此时，我们可以使用共享全局变量的方式来解决：

val data = Array(1,2,3,4,5)
    // 使用全局共享变量的累加器(accumulator)
    val counter = sc.accumulator(0)
    var rdd = sc.parallelize(data)
    rdd.foreach(x => counter+=x)
    println("counter的值：" + counter)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70 1]  
发现结果是对的，此时已经将 `counter`变量设置为共享全局变量。

Spark 有两种共享变量的方式：`广播变量（broadcast variable）`与`累加器（accumulator）`，此处我们使用的是累加器，下面还会详细讲解。

##### 3. 打印 RDD 的元素 #####

*  如果是 Spark 是单机模式，可以使用 `rdd.foreach(println)`或`rdd.map(println)`来打印 RDD 的元素。但是如果是集群模式，就需要考虑闭包问题了，因为此时，执行器输出写入的是`执行器`的`stdout`，而不是`驱动程序`的 `stdout`。所以如果那么要想在`驱动程序`中打印元素，可以怎么操作呢？

1.  如果数据量不多的话可以使用 `collect()`方法：

rdd.collect().foreach(println);

1.  如果数据量很大的话可以使用 `take()`方法：

rdd.take(num).foreach(println);

RDD 算子相关教程参考： [Spark RDD的实操教程（二）][Spark RDD]

# 0x03 共享变量 #

Spark 为了能够让多节点之间共享变量，提供了两种方法：广播变量和累加器。

##### 1. 广播变量 #####

广播变量可以在每个节点上缓存一个只读的变量，可以提高数据的分配效率。

1.  创建方式：

sc.broadcast(v)

可以在Spark Shell中执行：

// 创建广播变量
    val broadcastVar = sc.broadcast(Array(1, 2, 3))
    // 查看广播变量的值
    broadcastVar.value

![在这里插入图片描述][20200709213555955.png]

##### 2. 累加器 #####

累加器只能够通过`加操作`的变量，因此在并行操作中具备更高的效率，可以实现 `counters`和 `sums`。

可以在Spark Shell中执行：

// 创建累加器
    val accum = sc.accumulator(0, "Accumulator")
    // 对RDD数据集的数值进行累加
    sc.parallelize(Array(1, 2, 3)).foreach(x => accum += x)
    // 查看累加器的结果
    accum.value

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70 2]

# 0xFF 总结 #

1.  本文介绍了向Spark传递参数的两种方式，其实一般只会用匿名函数这样的用法；闭包的概念其实也比较简单，需要理解；共享变量的两种方式必须要掌握，包括原理。
2.  请继续学习相关课程，加油！

--------------------

**福利：**  
[邵奈一的技术博客导航][Link 6]

邵奈一原创不易，如转载请标明出处，教育是一生的事业。

--------------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70]: /images/20210923/b7c80d2c9a5f4e8db795824ad84feade.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70 1]: /images/20210923/f4812b3909c346ba9a4c36442681a07f.png
[Spark RDD]: https://blog.csdn.net/shaock2018/article/details/107204691
[20200709213555955.png]: /images/20210923/1fddff5af49b4db5a9cbc0cc01656801.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg_size_16_color_FFFFFF_t_70 2]: /images/20210923/8085b0ef5d05408d9a6b0aecb585091f.png
[Link 1]: https://blog.csdn.net/shaock2018
[Link 2]: https://img-blog.csdnimg.cn/20190608000417972.jpeg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg=,size_16,color_FFFFFF,t_70
[Link 3]: https://img-blog.csdnimg.cn/20190514084602748.png
[Link 4]: https://weibo.com/5680168183/
[Link 5]: https://www.jianshu.com/u/6f033c6a3b90
[Link 6]: https://blog.csdn.net/shaock2018/article/details/89349249