在 Spark 中，广播是如何工作的？

Question

这是一个非常简单的问题：在 spark 中，broadcast 可以用来高效地将变量发送给执行器。这是如何工作的？

更准确地说：

• 什么时候发送值：只要我打电话给broadcast，或者什么时候使用这些值？
• 数据到底发送到了哪里：发送给所有执行者，还是只发送给需要它的人？
• 数据存储在哪里？在内存中，还是在磁盘上？
• 简单变量和广播变量的访问方式有区别吗？当我调用 .value 方法时，底层会发生什么？

Answer 1

简答

• 值在执行程序第一次需要时发送。调用 sc.broadcast(variable) 时不会发送任何内容。
• 数据仅发送到包含需要它的执行程序的节点。
• 数据存储在内存中。如果没有足够的可用内存，则使用磁盘。
• 是的，访问局部变量和广播变量之间有很大的不同。广播变量必须在第一次访问时下载。

长答案

答案在 Spark 的源代码中，TorrentBroadcast.scala。

1. 当sc.broadcast 被调用时，一个新的TorrentBroadcast 对象将从BroadcastFactory.scala 实例化。在初始化 TorrentBroadcast 对象时调用的writeBlocks() 中发生以下情况：

   1. 对象使用MEMORY_AND_DISK 策略在本地缓存，未序列化。
   2. 已序列化。
   3. 序列化版本被拆分为 4Mb 块，这些块被压缩^[0]，并保存在本地^[1]。

2. 当新的执行器被创建时，它们只有轻量级的TorrentBroadcast 对象，它只包含广播对象的标识符，以及它的块数。

3. TorrentBroadcast 对象具有包含其值的惰性^[2] 属性。当调用value 方法时，会返回这个惰性属性。所以第一次在任务上调用这个值函数时，会发生以下情况：

   1. 以随机顺序从本地块管理器获取块并解压缩。
   2. 如果它们不在本地块管理器中，则在块管理器上调用getRemoteBytes 来获取它们。 网络流量发生仅在那个时候。
   3. 如果该块不存在于本地，则使用 MEMORY_AND_DISK_SER 对其进行缓存。

---

^[0] 默认使用 lz4 压缩。 This can be tuned.

^[1] 块存储在本地block manager，使用MEMORY_AND_DISK_SER，这意味着它将不适合内存的分区溢出到磁盘。每个块都有一个唯一标识符，由广播变量的标识符及其偏移量计算得出。块大小can be configured；默认为 4Mb。

^[2] scala 中的 lazy val 是一个变量，它的值在第一次访问时被评估，然后被缓存。 See the documentation.

Answer 2

• 一经播出
• 它使用 torrent 协议发送给所有执行器，但仅在需要时加载
• 一旦加载的变量被反序列化存储在内存中

• 它：

  • 验证广播没有被破坏
  • 从 blockManager 中延迟加载变量