讲共享变量如何在数据管道使用中间结果海量资源.pdfVIP

本文由简悦SimpRead转码，原文地址

今天我将为你讲解：如何用共享变量在数据管道中使用中间结果。共享变量是Spark中进阶特性之一，

一共有两种：

广播变量；

累加器。

这两种变量可以认为是在用算子定义的数据管道外的两个全局变量，供所有计算任务使用。在Spark作

业中，用户编写的高阶函数会在集群中的Executor里执行，这些Executor可能会用到相同的变量，这

些变量被到每个Executor中，而Executor对变量的更新不会传回Driver。

在计算任务中支持通用的可读写变量一般是低效的，即便如此，Spark还是提供了两类共享变量：广播

变量（broadcastvariable）与累加器（accumulator）。当然，对于分布式变量，如果不加限制会出

现一致性的问题，所以共享变量是两种非常特殊的变量。

广播变量：只读；

累加器：只能增加。

广播变量

广播变量类似于MapReduce中的DistributeFile，通常来说是一份不大的数据集，一旦广播变量在

Driver中被创建，整个数据集就会在集群中进行广播，能让所有正在运行的计算任务以只读方式。

广播变量支持一些简单的数据类型，如整型、集合类型等，也支持很多复杂数据类型，如一些自定义的

数据类型。

广播变量为了保证数据被广播到所有节点，使用了很多办法。这其实是一个很重要的问题，我们不能期

望100个或者1000个Executor去连接Driver，并拉取数据，这会让Driver重负。Executor采

用的是通过HTTP连接去拉取数据，类似于BitTorrent点对点传输。这样的方式更具扩展性，避免了所

有Executor都去向Driver请求数据而造成Driver故障。

Spark广播机制运作方式是这样的：Driver将已序列化的数据切分成小块，然后将其在自己的块管

理器BlockManager中，当Executor开始运行时，每个Executor首先从自己的块管理器中试图

获取广播变量，如果以前广播过，那么直接使用；如果没有，Executor就会从Driver或者其他可用的

Executor去拉取数据块。一旦拿到数据块，就会放到自己的块管理器中。供自己和其他需要拉取的

Executor使用。这就很好地防止了Driver单点的性能瓶颈，如下图所示。

下面来看看如何在Spark作业中创建、使用广播变量。代码如下：

scalavalrdd_one=sc.parallelize(Seq(1,2,3))

rdd_one:org.apache.spark.rdd.RDD[Int]=ParallelCollectionRDD[101]at

parallelizeatconsole:25

scalavali=5

i:Int=5

scalavalbi=sc.broadcast(i)

bi:org.apache.spark.broadcast.Broadcast[Int]=Broadcast(147)

scalabi.value

res166:Int=5

scalardd_one.take(5)

res164:Array[Int]=Array(1,2,3)

scalardd_one.map(j=j+bi.value).take(5)

res165:Array[Int]=Array(6,7,8)

在用户定义的高阶函数中，可以直接使用广播变量的。下面看一个集合类型的广播变量：

scalavalrdd_one=sc.parallelize(Seq(1,2,3))

rdd_one:org.apache.spark.rdd.RDD[Int]=ParallelCollectionRDD[109]at

parallelizeatconsole:25

scalavalm=scala.collection.mutable.HashMap(1-2,2-3,3-4)

m:scala.collection.mutable.HashMap[Int,Int]=Map(