广播变量(Broadcast)-及生命周期

1 Spark中广播变量概念

广播变量是spark中共享变量的其中一种。它可以让程序高效的向所有工作节点发送一个只读的值，以供一个或多个spark操作使用。

2 广播变量使用示例

下面是一段简单的计算逻辑，对比了不使用广播变量和使用广播变量的两种方式：

val pws = Map("Apache Spark" -> "http://spark.apache.org/", "Scala" -> "http://www.scala-lang.org/")

//不使用广播变量（比较低效，需要多次将pws变量发送到各个节点）

val time1=System.currentTimeMillis()

val websites = sc.parallelize(Seq("Apache Spark", "Scala")).map(pws).collect

val time2 =System.currentTimeMillis()

println("websites = " + (time2 - time1))

websites.foreach(println)

//使用广播变量

val pwsB = sc.broadcast(pws)

val time3=System.currentTimeMillis()

val websitesWithBroadcast = sc.parallelize(Seq("Apache Spark", "Scala")).map(pwsB.value).collect

println("websitesWithBroadcast = " + (System.currentTimeMillis() - time3))

websitesWithBroadcast.foreach(println)

输出：

===============广播变量使用示例================

cost = 446ms

websites =

http://spark.apache.org/ http://www.scala-lang.org/

cost = 32ms

websitesWithBroadcast =

可以看出，输出结果是一样的，但是耗时，广播变量明显优于非广播变量模式。

3 广播变量原理

将变量广播到各个执行器：

在spark中，使用SparkContext创建广播变量，使用broadcastmanager和ContextCleaner管理其生命周期。

org.apache.spark.SparkContext#broadcast核心代码:

def broadcast

T: ClassTag

: Broadcast[T] = {

//调用broadcastManager创建新的广播变量

val bc = env.broadcastManager.newBroadcast

T

//注册广播变量到ContextCleaner

cleaner.foreach(_.registerBroadcastForCleanup(bc))

bc

}

也可以手动销毁广播变量org.apache.spark.broadcast.Broadcast#destroy:

def destroy() {

destroy(blocking = true)

获取广播变量的值:

/* Get the broadcasted value. /

def value: T = {

assertValid()

getValue()

转载地址

http://xiajunhust.github.io/2018/12/23/Spark%E5%9F%BA%E7%A1%80%E5%AD%A6%E4%B9%A0%E7%AC%94%E8%AE%B03-%E5%B9%BF%E6%92%AD%E5%8F%98%E9%87%8F/