Spark中广播变量详解

【前言:Spark目前提供了两种有限定类型的共享变量:广播变量和累加器,今天主要介绍一下基于Spark2.4版本的广播变量。先前的版本比如Spark2.1之前的广播变量有两种实现:HttpBroadcast和TorrentBroadcast,但是鉴于HttpBroadcast有各种弊端,目前已经舍弃这种实现,本篇文章也主要阐述TorrentBroadcast】

广播变量概述

广播变量是一个只读变量,通过它我们可以将一些共享数据集或者大变量缓存在Spark集群中的各个机器上而不用每个task都需要copy一个副本,后续计算可以重复使用,减少了数据传输时网络带宽的使用,提高效率。相比于Hadoop的分布式缓存,广播的内容可以跨作业共享。

广播变量要求广播的数据不可变、不能太大但也不能太小(一般几十M以上)、可被序列化和反序列化、并且必须在driver端声明广播变量,适用于广播多个stage公用的数据,存储级别目前是MEMORY_AND_DISK。

广播变量存储目前基于Spark实现的BlockManager分布式存储系统,Spark中的shuffle数据、加载HDFS数据时切分过来的block块都存储在BlockManager中,不是今天的讨论点,这里先不做详述了。

广播变量的创建方式和获取

//创建广播变量

val broadcastVar = sparkSession.sparkContext.broadcast(Array(1, 2, 3))

//获取广播变量
broadcastVar.value

广播变量实例化过程

1.首先调用val broadcastVar = sparkSession.sparkContext.broadcast(Array(1, 2, 3))

2.调用BroadcastManager的newBroadcast方法

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)

4.在构建TorrentBroadcast时,将广播的数据写入BlockManager

1)首先会将广播变量序列化后的对象划分为多个block块,存储在driver端的BlockManager,这样运行在driver端的task就不用创建广播变量的副本了(具体可以查看TorrentBroadcast的writeBlocks方法)

2)每个executor在获取广播变量时首先从本地的BlockManager获取。获取不到就会从driver或者其他的executor上获取,获取之后,会将获取到的数据保存在自己的BlockManager中

3)块的大小默认4M

conf.getSizeAsKb("spark.broadcast.blockSize", "4m").toInt * 1024

广播变量初始化过程

1.首先调用broadcastVar.value

2.TorrentBroadcast中lazy变量_value进行初始化,调用readBroadcastBlock()

3.先从缓存中读取,对结果进行模式匹配,匹配成功的直接返回

4.读取不到通过readBlocks()进行读取
从driver端或者其他的executor中读取,将读取的对象存储到本地,并存于缓存中

new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK)

Spark两种广播变量对比

正如【前言】中所说,HttpBroadcast在Spark后续的版本中已经被废弃,但考虑到部分公司用的Spark版本较低,面试中仍有可能问到两种实现的相关问题,这里简单介绍一下:

HttpBroadcast会在driver端的BlockManager里面存储广播变量对象,并且将该广播变量序列化写入文件中去。所有获取广播数据请求都在driver端,所以存在单点故障和网络IO性能问题。

TorrentBroadcast会在driver端的BlockManager里面存储广播变量对象,并将广播对象分割成若干序列化block块(默认4M),存储于BlockManager。小的block存储位置信息,存储于Driver端的BlockManagerMaster。数据请求并非集中于driver端,避免了单点故障和driver端网络磁盘IO过高。

TorrentBroadcast在executor端存储一个对象的同时会将获取的block存储于BlockManager,并向driver端的BlockManager汇报block的存储信息。
请求数据的时候会先获取block的所有存储位置信息,并且是随机的在所有存储了该executor的BlockManager去获取,避免了数据请求服务集中于一点。

总之就是HttpBroadcast导致获取广播变量的请求集中于driver端,容易引起driver端单点故障,网络IO过高影响性能等问题,而TorrentBroadcast获取广播变量的请求服务即可以请求到driver端也可以在executor,避免了上述问题,当然这只是主要的优化点。

上一篇:Spark RDD详解 —— RDD特性、lineage、缓存、checkpoint、依赖关系


下一篇:《高效学习OpenGL》之缓冲区对象 glGenBuffers(), glBindBuffer(), glBufferData(),glMapBuffer()