java - 在多个 Spark Executor 上共享 Zookeeper 配置

Question

我有一个用Zookeeper写的配置信息。我正在使用 Apache Curator 和 Curator Watcher 来读取配置(如果有更好的解决方案来读取它，我很乐意使用它)，因此如果 Zookeeper 中的配置发生更改，我将收到新的配置。我在 Spark 中使用此配置。如何将其共享给同一应用程序的所有 Spark 执行器？

谢谢!

LE:

谢谢迪凯，

在下面的代码中，您将在哪里执行观察者实现？我是 Spark 新手，不太确定每个工作人员会做什么。

谢谢!

final JavaDStream<ElementMessage> nodeMessageStream = mapWithStateDistinctAndFiltered.flatMap(pair -> pair._2.buildElementMessages())
            .filter(f -> f != null);

    nodeMessageStream.foreachRDD(rdd -> {
        rdd.foreachPartition(r -> {
            final ElementRecordRestClient rest = new ElementRecordRestClient(
                    startProps.getProperty(InputPropertyKey.WEPAPP_URL.toString()));
            r.forEachRemaining(message -> {
                rest.createObject(message.toElementRecord());
            });
        });
    });

Answer 1

在这种情况下，我要做的是在主节点上运行 Curator Watcher，并使用 Spark 的广播变量将配置广播到所有执行器。每当配置发生更改时，您都会停止当前的流上下文，并使用新配置启动一个新的流上下文。这将确保您的结果始终一致。

另一种方法是读取foreachPartition内的zookeeper配置 lambda 函数。但由于配置是由每个分区独立读取的，因此同一 RDD 的不同分区可以获得不同的配置，这可能不是您所期望的。