Spark广播变量分析以及如何动态更新广播变量
本篇文章给大家分享的是有关Spark广播变量分析以及如何动态更新广播变量，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。
你所需要的网站建设服务，我们均能行业靠前的水平为你提供.标准是产品质量的保证，主要从事成都网站设计、做网站、企业网站建设、手机网站制作设计、网页设计、成都品牌网站建设、网页制作、做网站、建网站。创新互联拥有实力坚强的技术研发团队及素养的视觉设计专才。
广播变量存储目前基于Spark实现的BlockManager分布式存储系统，Spark中的shuffle数据、加载HDFS数据时切分过来的block块都存储在BlockManager中，不是今天的讨论点，这里先不做详述了。
广播变量的创建方式和获取
//创建广播变量
val broadcastVar = sparkSession.sparkContext.broadcast(Array(1, 2, 3))
//获取广播变量
broadcastVar.value1.首先调用val broadcastVar = sparkSession.sparkContext.broadcast(Array(1, 2, 3))
val bc = env.broadcastManager.newBroadcast[T](value, isLocal)
3.通过广播工厂的newBroadcast方法进行创建
broadcastFactory.newBroadcast[T](value_, isLocal, nextBroadcastId.getAndIncrement())
在调用BroadcastManager的newBroadcast方法时已完成对广播工厂的初始化（initialize方法），我们只需看BroadcastFactory的实现TorrentBroadcastFactory中对TorrentBroadcast的实例化过程：
new TorrentBroadcast[T](value_, id)
conf.getSizeAsKb("spark.broadcast.blockSize", "4m").toInt * 1024广播变量初始化过程
从driver端或者其他的executor中读取，将读取的对象存储到本地，并存于缓存中
new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK)
Spark两种广播变量对比
TorrentBroadcast在executor端存储一个对象的同时会将获取的block存储于BlockManager，并向driver端的BlockManager汇报block的存储信息。
总之就是HttpBroadcast导致获取广播变量的请求集中于driver端，容易引起driver端单点故障，网络IO过高影响性能等问题，而TorrentBroadcast获取广播变量的请求服务即可以请求到driver端也可以在executor，避免了上述问题，当然这只是主要的优化点。
既然无法更新，那么只能动态生成，应用场景有实时风控中根据业务情况调整规则库、实时日志ETL服务中获取最新的日志格式以及字段变更等。
@volatile private var instance: Broadcast[Array[Int]] = null
//获取广播变量单例对象
def getInstance(sc: SparkContext, ctime: Long): Broadcast[Array[Int]] = {
if (instance == null) {
synchronized {
if (instance == null) {
instance = sc.broadcast(fetchLastestData())
}
}
}
instance
}
//加载要广播的数据，并更新广播变量
def updateBroadCastVar(sc: SparkContext, blocking: Boolean = false): Unit = {
if (instance != null) {
//删除缓存在executors上的广播副本，并可选择是否在删除完成后进行block等待
//底层可选择是否将driver端的广播副本也删除
instance.unpersist(blocking)
instance = sc.broadcast(fetchLastestData())
}
}
def fetchLastestData() = {
//动态获取需要更新的数据
//这里是伪代码
Array(1, 2, 3)
}val dataFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")
...
...
stream.foreachRDD { rdd =>
val current_time = dataFormat.format(new Date())
val new_time = current_time.substring(14, 16).toLong
//每10分钟更新一次
if (new_time % 10 == 0) {
updateBroadCastVar(rdd.sparkContext, true)
}
rdd.foreachPartition { records =>
instance.value
...
}
}Spark流式程序中为何使用单例模式
1.广播变量是只读的，使用单例模式可以减少Spark流式程序中每次job生成执行，频繁创建广播变量带来的开销
2.广播变量单例模式也需要做同步处理。在FIFO调度模式下，基本不会发生并发问题。但是如果你改变了调度模式，如采用公平调度模式，同时设置Spark流式程序并行执行的job数大于1，如设置参数spark.streaming.concurrentJobs=4，则必须加上同步代码
3.在多个输出流共享广播变量的情况下，同时配置了公平调度模式，也会产生并发问题。建议在foreachRDD或者transform中使用局部变量进行广播，避免在公平调度模式下不同job之间产生影响。
除了广播变量，累加器也是一样。在Spark流式组件如Spark Streaming底层，每个输出流都会产生一个job，形成一个job集合提交到线程池里并发执行。
以上就是Spark广播变量分析以及如何动态更新广播变量，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注创新互联行业资讯频道。
标题名称：Spark广播变量分析以及如何动态更新广播变量
标题URL：http://cdiso.cn/article/ihgcio.html
