在 Spark 中，广播变量（Broadcast Variables） 是一种特殊类型的共享变量，用于高效地在集群中的所有节点间分发大型只读数据集。它解决了 Spark 任务中频繁传输重复数据的性能问题，特别适用于需要在多个任务中重用相同数据的场景。

为什么需要广播变量？

在 Spark 中，当一个函数（如 map()、filter()）引用了驱动程序（Driver）中的变量时，Spark 会默认将该变量的副本发送给每个任务（Task）。如果变量很大（例如，一个包含百万条记录的 lookup 表）：

会导致大量网络传输，浪费带宽
消耗每个 Executor 的内存
降低任务执行效率

广播变量通过一次将数据分发到每个节点（而非每个任务），并在节点上缓存数据，避免了重复传输和存储，显著提升性能。

广播变量的核心特性

只读性：一旦广播变量被创建，就不能被修改（保证数据一致性）。
节点级缓存：每个工作节点（Worker Node）只会存储一份广播变量的副本，供该节点上的所有任务共享。
高效分发：Spark 使用 P2P 协议（BitTorrent 类似机制） 分发大型广播变量，避免 Driver 成为瓶颈。
惰性评估：广播变量在第一次被任务使用时才会被实际分发到节点。

使用场景

大型查找表（例如，将 ID 映射到名称的字典）
机器学习模型参数（如训练好的权重矩阵）
配置文件或常量数据集
需要在多个转换操作中重用的大型数据结构

如何使用广播变量？

广播变量的使用步骤如下：

创建广播变量：通过 SparkContext.broadcast(value) 方法，将驱动程序中的变量封装为广播变量。
在任务中使用：通过 .value 属性访问广播变量的值（在 Executor 中）。
销毁广播变量（可选）：通过 .unpersist() 方法释放节点上的缓存，或 .destroy() 彻底销毁变量。

示例代码（Scala）

import org.apache.spark.sql.SparkSessionobject BroadcastExample {def main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("BroadcastExample").master("local[*]") // 本地模式，实际生产环境不需要.getOrCreate()val sc = spark.sparkContext// 1. 定义一个大型数据集（例如，ID到名称的映射）val largeLookupTable = Map(1 -> "Alice",2 -> "Bob",3 -> "Charlie",// ... 假设包含百万条记录)// 2. 创建广播变量val broadcastVar = sc.broadcast(largeLookupTable)// 3. 创建一个RDD（例如，包含ID的数据集）val idsRDD = sc.parallelize(Seq(1, 2, 3, 1, 2))// 4. 在任务中使用广播变量（通过.value访问）val namesRDD = idsRDD.map(id => broadcastVar.value.getOrElse(id, "Unknown"))// 输出结果namesRDD.collect().foreach(println)// 输出：Alice, Bob, Charlie, Alice, Bob// 5. 销毁广播变量（释放资源）broadcastVar.unpersist()spark.stop()}
}

示例代码（Python）

from pyspark.sql import SparkSessionif __name__ == "__main__":spark = SparkSession.builder \.appName("BroadcastExample") \.master("local[*]") \.getOrCreate()sc = spark.sparkContext# 1. 定义大型查找表large_lookup_table = {1: "Alice",2: "Bob",3: "Charlie"# ... 假设包含百万条记录}# 2. 创建广播变量broadcast_var = sc.broadcast(large_lookup_table)# 3. 创建ID的RDDids_rdd = sc.parallelize([1, 2, 3, 1, 2])# 4. 使用广播变量names_rdd = ids_rdd.map(lambda id: broadcast_var.value.get(id, "Unknown"))# 输出结果print(names_rdd.collect())  # ['Alice', 'Bob', 'Charlie', 'Alice', 'Bob']# 5. 释放资源broadcast_var.unpersist()spark.stop()