当前位置：首页 > article >正文

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

article 2024/10/28 3:13:43

在 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 程序性能至关重要。开发者需要注意控制广播数据的大小，合理管理其生命周期，并选择合适的序列化器来提升性能。

查看全文

http://www.kler.cn/news/368137.html