接着之前的Spark调优系列文章，我们今天介绍一下通过广播的方式优化我们的Spark作业运行效率。在介绍文章之前，我们首先来分析一下我们Spark作业运行的时候每个task任务默认是怎么怎么工作的。好了，首先先来看一张图。
默认task的工作流程图
上图是用户session分析模块中的按照时间比例随机抽取的session信息的集合（解释：这里是结合我项目中的一个点进行说明的，举一反三，在以后的项目中遇到类似的情况就可以参考这个例子）。上图中Task执行一个算子的时候，使用了“随机抽取数据的map”这个外部变量，默认情况下，每个task都会获取到这个外部变量的一份副本，那么这样做有什么缺点呢？在数据量增大的情况下，我们的性能会收到什么样的影响呢？围绕这两个问题我们开始下面的讨论。
这个map结构存放的数据是一个个的Entry实体等，所以map整体上相对来说是比较耗费内存的。从网络传输和内存占用两个方面进行简单的分析一下；
一，说说网络传输方面。比如这个map占用的内存是1M（事实上，当数据量比较大的时候，这个map肯能占用的内存远远大于1M），然后在这个Spark作业中，我们假定运行了1000个task（为了程序并行运行，加快程序的处理速度）。每个task都会运行一个map副本，导致map会被拷贝1000份，然后通过网络传输到各个task中去使用。很显然这样就会又1G的数据通过网络进行传输，想象一下，如果map在大些的话，后果将更加不堪设想呐！！！仅仅一个map的网络传输就会消耗不少的集群性能。
二，说说内存消耗方面。map副本传输到各个task上之后，1M的map有1000个分布在集群上，瞬间消耗1G的内存。后果很容易想象，大量的数据对象占用内存，很容易触发JVM的GC，一旦发生GC，Spark作业就会停下来等待。频繁的GC会对Spark作业的运行速度造成不小的影响。
通过以上的简单分析，对于一些外部的数据在一个算子中使用的时候（本例中的随机抽取数据的map就是一个典型的例子），我们必须要想一些办法，一方面减少网络IO造成的影响，另一方面减少内存的占用。Spark作为一个内存计算框架当然已经考虑到这种情况了，那就是通过广播变量的方式来优化以上提出的两个问题。那么什么是广播变量呢？
广播变量：在程序初始化的时候，会把数据在Driver上面存放一个副本（本例中就对应的就是随机抽取数据的map）。task在运行的时候，在使用广播变量中的数据，此时首先会在Executor对应的BlockManager中尝试获取变量副本；如果本地内存没有，那就会从Driver远程拷贝一份过来，并存放在本地的BlockManager中，以后在使用该数据时，直接在BlockManager中获取就可以了，从而减少网络IO和内存的占用。executor的BlockManager除了会去从Driver中获取副本，也会从就近的其他节点上去查找副本，说通俗一点就是就近获取。
BlockManager：BlockManager是负责管理每个Executor上对应的内存和磁盘上的数据，每个Executor上都有一个BlockManager。
下面用一幅图来总结一下上面所说的执行流程。
广播变量执行流程图
在附上基于Java的Spark部分实现程序，如下图所示
包装成Broadcast
使用包装后的Broadcast
以上即为通过使用广播变量的方式降低网络IO和内存占用，如有没有将明白的，欢迎大家留言。本文到这里基本接近尾声，后续还会不断更新关于Spark作业优化的一些其他方式，欢迎关注。
如需转载，请注明：
网友评论