apache-spark - 广播变量什么时候可以改变？

Question

有人告诉我广播变量应该是不可变的。

然而，我看到了一个代码片段，其中广播变量用作标志。

public class TestBroadcast {

  private static JavaStreamingContext jssc;
  private static volatile Broadcast<Boolean> done;

  public static void main(String[] args) throws InterruptedException {

    Logger.getLogger("org").setLevel(Level.ERROR);

    List<String> log = Arrays.asList("X", "X", "X");

    SparkConf sparkConf = new SparkConf().setAppName("Test").setMaster("local[2]");
    jssc = new JavaStreamingContext(sparkConf, Durations.seconds(1));
    done = jssc.sparkContext().broadcast(Boolean.FALSE);    // false in the beginning

    JavaRDD<String> _rdd = jssc.sparkContext().parallelize(log);
    Queue<JavaRDD<String>> queue = new LinkedList<>();
    queue.add(_rdd);
    JavaDStream<String> lines = jssc.queueStream(queue);

    lines.foreachRDD(
        rdd -> {
          rdd.foreachPartition(x -> System.out.println(done.getValue())); // executor get false
          done = jssc.sparkContext().broadcast(Boolean.TRUE); // driver set the variable to true
/*MARK*/  rdd.foreachPartition(x -> System.out.println(done.getValue())); // executor get true
        });

    jssc.start();

    jssc.awaitTermination();

  }

}

广播变量在/*MARK*/(original source)注释的行发生变化，为什么会这样？

Answer 1

尽管 done 名称相同，但这两个广播变量是不同的。

我必须承认，我从未见过这样使用广播变量(可能是因为它会导致错误的结论，正如您的问题似乎已经证明的那样)。除非我弄错了，否则使用不会真正增加太多，因为 bool 值非常小(即使没有广播变量也不会增加序列化消息有效负载)。

更不寻常的是，这是在一个 Spark Streaming 应用程序中，该应用程序使用驱动程序上发生的 foreachRDD，因此可以访问本来不可用的 JavaStreamingContext在执行者上(会导致 NullPointerException)。

---

事实上，广播变量的生命周期允许改变广播变量的值。将广播变量视为执行程序内存空间中可用的某些内容的句柄，可以按需解析为值。

你可以使用 unpersist驱动程序上的方法(这将触发“执行者删除”消息发送给执行者)，因此以下 value 将再次获得广播值(可能与最初的值不同)。

unpersist(): Unit Asynchronously delete cached copies of this broadcast on the executors. If the broadcast is used after this is called, it will need to be re-sent to each executor.

对于这种特殊情况(使用 foreachRDD)，它没有多大意义，因为 foreachRDD 引入了另一个层，您可以在其中操作广播变量并提交 Spark 作业。