在 Spark 中定期更新全局变量

Question

我在 spark 流应用中做类似模式匹配的事情。我想要的是更新像 broadcast variable 这样的变量，但是它是可变的。有没有办法做到这一点？有什么建议吗？

编辑

抱歉没有说清楚。我在日志上做一些 CEP 的东西。我需要在 spark 应用程序运行时从 elasticsearch 加载规则。我想将这些规则应用到工作人员方面（在每个 RDD 上）。

Answer 1

这里的想法是为定期刷新的广播变量编写一个包装器。 Catch 是在允许 RDD-RDD 操作的转换（或任何其他变体）中调用此函数。

为 BroadcastWrapper 类截取的代码：

public class BroadcastWrapper {

private Broadcast<ReferenceData> broadcastVar;
private Date lastUpdatedAt = Calendar.getInstance().getTime();

private static BroadcastWrapper obj = new BroadcastWrapper();

private BroadcastWrapper(){}

public static BroadcastWrapper getInstance() {
    return obj;
}

public JavaSparkContext getSparkContext(SparkContext sc) {
   JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc);
   return jsc;
}

public Broadcast<ReferenceData> updateAndGet(SparkContext sparkContext){
    Date currentDate = Calendar.getInstance().getTime();
    long diff = currentDate.getTime()-lastUpdatedAt.getTime();
    if (var == null || diff > 60000) { //Lets say we want to refresh every 1 min = 60000 ms
        if (var != null)
           var.unpersist();
        lastUpdatedAt = new Date(System.currentTimeMillis());

        //Your logic to refresh
        ReferenceData data = getRefData();

        var = getSparkContext(sparkContext).broadcast(data);
   }
   return var;
 }
}

要使用此方法，我们可以执行以下操作 ->

objectStream.transform(stream -> {
    Broadcast<ReferenceData> refdataBroadcast = BroadcastWrapper.getInstance().updateAndGet(stream.context());

 /*Your code here*/

});

请在另一个线程上查看我的答案以获得更好的清晰度 https://stackoverflow.com/a/41259333/3166245