spark broadcast

2023年4月20日 246次阅读 来源: 丧诗
  1. 为什么需要broadcast类型变量 ,它有哪些优点?
  2. spark中怎样创建和使用broadcast类型变量 ?
  3. spark中的具体实现

1. 为什么需要broadcast类型变量 ?

各个slave端都需要同一个数据,并且只有读取操作
例如: 一个object对象,一个map或者bloomFilter等

broadcast类型变量和传输一个可序列化的变量的区别 ?

  1. broadcast类型变量可以保证只在executor的内存中存在一份
  2. 将要传输的变量不需要实现Serializable接口
  3. 可以高效地传输较大的数据集

以上3点可以在下面的实现中看到

2. spark中怎样创建和使用broadcast ?

具体的示例在Broadcast.scala的comment中,其中需要注意的是,一旦一个broadcast初始化好了,今后对它的值的访问只能通过broadcast间接访问,这也是一个wrapper的使用模式,例如:

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)
scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)

3. spark中的具体实现

1. broadcast类型具体的方法

1. 创建

sc.broadcast()

2. 读取

value

3. 销毁

  1. 可重建的销毁: unpersist
  2. 不可重建的销毁: destroy

2. 实现机制

目前spark中只有一种实现 TorrentBroadcast.scala

具体机制如下:
driver端:

将序列化过的对象分成小块,存放在driver端的BlockManager里

executor端:

  1. executor首先从自己的BlockManager去拿,如果有就直接用,如果没有执行2步
  2. 从driver/其它executor端拉取对象的小块放入自己的BlockManager 供自己和其它的executor使用/拉取

这样的好处是避免了driver端因为发送数据给每个executor而造成热点问题

TorrentBroadcast 实现摘要

TorrentBroadcast#value调用顺序:

  1. getValue 外部接口
  2. readBroadcastBlock 将读入的blocks拼成obj放入blockManager
  3. readBlocks 将块读取入blockManager

broadcast变量的块的形成:
只会在driver端形成,调用顺序

  1. private val numBlocks: Int = writeBlocks(obj)
  2. writeBlocks 将对象分成块

executor端也会有取numBlocks这个变量,但由于不是transient的,就不会再调用writeBlocks

ps:
writeBlocks: 将对象切分为块,返回块的个数
每块的id为: broadcast_1_piece2 (其中1表示broadcast的id,2表示第几个块),由于数据是在driver端收集的,然后其他executor通过byte流来拉取数据,这样就不需要broadcast的变量实现Serializable接口

    原文作者：丧诗
    原文地址: https://www.jianshu.com/p/c5c71bdcccc0
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞