Spark 广播变量（Broadcast Variable）原理及源码分析

    在 **Apache Spark** 中，**广播变量（Broadcast Variable）**是用于优化**跨节点共享数据**的一种机制。广播变量的目标是避免多个任务重复传输相同的数据，提高执行效率。它通过将数据只发送一次，并缓存到每个执行器节点，从而减少通信开销。

    在这篇文章中，我们将详细全面地从底层原理、执行流程和源码角度解释 Spark 的广播变量工作机制。

1. 广播变量的背景和需求

    在 Spark 应用中，有时需要在所有任务中使用相同的只读数据（如字典表、机器学习模型等）。如果不使用广播变量，这些数据会被每个任务各自传输到节点，这会产生大量的网络传输开销。

广播变量的目标：

将大数据只传输一次，避免任务间的重复传输。
在每个 Executor 上缓存广播数据，任务可以直接从本地获取。
保证广播的数据在整个计算过程中只读。

2. Spark 广播变量的核心原理

2.1 广播机制的基本概念

Driver 广播数据：Driver 将要广播的数据生成广播变量，并通过 Spark 内部的广播机制将数据传输到所有 Executor。
Executor 缓存数据：每个 Executor 将接收到的数据缓存到本地内存或磁盘中。

任务读取广播数据：任务在执行过程中从本地的缓存中获取广播的数据，而不需要再通过网络传输。

 Spark 的广播机制基于一个**高效的点对点传输协议**，实现了类似于 **BitTorrent** 的数据传播方式，以减少网络带宽的占用。

3. 广播变量的生命周期与执行流程

广播变量的生命周期可以分为以下几个阶段：

创建广播变量（Driver 端）：使用 sc.broadcast() 在 Driver 中创建广播变量，序列化数据并启动广播。
将广播数据传递到 Executor：广播机制采用分层传播。Driver 首先将广播的数据块传输到一部分 Executor，这些 Executor 再将数据块传递给其他节点（类似于 BitTorrent 协议的文件传输）。
任务访问广播数据（Executor 端）：当任务运行时，如果发现 Executor 上已经存在所需的广播数据，就直接从本地读取；如果没有，则尝试通过网络获取。
销毁广播变量：在计算完成后，可以通过 unpersist() 手动移除广播变量的数据，释放内存。

4. Spark 广播变量的源码分析

**4.1 广播变量的核心类：

Broadcast

Broadcast

是一个抽象类，Spark 提供了两种广播实现：

**TorrentBroadcast**：基于 BitTorrent 协议实现的分块广播，适用于大数据量的场景。
**HttpBroadcast**：基于 HTTP 的广播，适用于小数据量场景。

我们来看

Broadcast

类的定义：

abstract class Broadcast[T](val id: Long) extends Serializable {
  def value: T
  def unpersist(blocking: Boolean = true): Unit
  def destroy(blocking: Boolean = true): Unit
}

**id**：广播变量的唯一标识。
**value**：获取广播变量的值，通常会从本地缓存中读取。
**unpersist()**：手动清理广播变量的数据。
**destroy()**：销毁广播变量。

**4.2 广播实现类：

TorrentBroadcast

TorrentBroadcast

是 Spark 默认的广播实现，它通过将广播数据拆分为多个块（blocks），然后在节点之间进行块级别的分发，类似于 BitTorrent 协议。

**

TorrentBroadcast

的构造方法**

class TorrentBroadcast[T: ClassTag](
    obj: T,
    id: Long,
    @transient private val sc: SparkContext)
  extends Broadcast[T](id) {

  // 将广播数据进行序列化并分块
  private val blocks: Array[ByteBuffer] = TorrentBroadcast.blockify(obj)

  // 启动广播
  TorrentBroadcast.writeBlocks(id, blocks, sc)
}

**blockify()**：将广播数据序列化，并分割为多个数据块。
**writeBlocks()**：将分割后的数据块传输到 Executor。

4.3 数据块的分发机制

TorrentBroadcast

使用了 Spark 的 BlockManager 组件来管理广播的数据块。每个数据块会通过 BlockManager 注册到本地存储，并通过网络传输给其他节点。

**

writeBlocks()

源码**

private def writeBlocks(id: Long, blocks: Array[ByteBuffer], sc: SparkContext): Unit = {
  val blockManager = sc.env.blockManager
  blocks.zipWithIndex.foreach { case (block, index) =>
    val blockId = BroadcastBlockId(id, index)
    blockManager.putSingle(blockId, block, StorageLevel.MEMORY_AND_DISK)
  }
}

**blockManager.putSingle()**：将每个数据块存储在本地的内存或磁盘上。

4.4 广播数据的读取

    当任务需要使用广播数据时，会通过 BlockManager 检查本地是否已有该数据块。如果没有，就会请求其他节点传输数据。

数据读取逻辑

override def value: T = {
  if (blocks == null) {
    readBlocks()  // 如果本地没有数据块，则从其他节点获取
  }
  TorrentBroadcast.unBlockify(blocks)
}

**readBlocks()**：尝试从其他节点获取缺失的数据块。
**unBlockify()**：将分块的数据重组为完整的数据。

5. 广播变量的持久化与销毁

    Spark 支持将广播变量的数据持久化到内存和磁盘中，并且可以通过

unpersist()

或

destroy()

来释放资源。

**unpersist()**：释放广播数据，但不删除元数据，可以在之后再次使用时重新加载。
**destroy()**：彻底销毁广播变量，并删除所有相关数据。

broadcastVar.unpersist()  // 手动释放资源
broadcastVar.destroy()    // 彻底销毁广播变量

6. 广播变量的优化与注意事项

数据序列化优化：广播的数据需要序列化，因此选择高效的序列化器（如 Kryo）可以提升性能。
数据大小控制：广播的数据不宜过大，否则可能导致内存不足。建议将数据量控制在 2GB 以下。
缓存机制：合理使用 unpersist() 释放不再需要的广播数据，避免内存泄漏。

7. 总结

    Spark 广播变量通过**序列化、分块、缓存和分层传播**的机制，大大减少了任务之间的数据传输开销，提高了性能。其核心实现基于

TorrentBroadcast

，利用了 BitTorrent 协议的思想进行数据的高效传播。通过源码分析，我们可以看到 Spark 广播变量如何通过 BlockManager 实现数据的分发和缓存，从而在大规模分布式计算中保持高效和稳定。

广播变量的正确使用对于优化 Spark 程序性能至关重要。开发者需要注意控制广播数据的大小，合理管理其生命周期，并选择合适的序列化器来提升性能。

标签： spark 大数据分布式

本文转载自: https://blog.csdn.net/goTsHgo/article/details/143116507
版权归原作者 goTsHgo 所有，如有侵权，请联系我们删除。