阿帕奇星火合并updateStateByKey（）

Question

我试图合并两个流，其中一人应该是有状态（如使用不频繁更新的静态数据）后：

SparkConf conf = new SparkConf().setAppName("Test Application").setMaster("local[*]"); 
JavaStreamingContext context = new JavaStreamingContext(conf, Durations.seconds(10)); 
context.checkpoint("."); 
JavaDStream<String> dataStream = context.socketTextStream("localhost", 9998); 
JavaDStream<String> refDataStream = context.socketTextStream("localhost", 9999); 

JavaPairDStream<String, String> pairDataStream = dataStream.mapToPair(e -> { 
    String[] tmp = e.split(" "); 
    return new Tuple2<>(tmp[0], tmp[1]); 
}); 

JavaPairDStream<String, String> pairRefDataStream = refDataStream.mapToPair(e -> { 
    String[] tmp = e.split(" "); 
    return new Tuple2<>(tmp[0], tmp[1]); 
}).updateStateByKey((Function2<List<String>, Optional<String>, Optional<String>>) (strings, stringOptional) -> { 
    if (!strings.isEmpty()) { 
     return Optional.of(strings.get(0)); 
    } 
    return Optional.absent(); 
}); 

pairDataStream.join(pairRefDataStream).print(); 


context.start(); 
context.awaitTermination(); 

当我写1 aaa到第一个流和1 111立即进入第二，一切正常，我看到合并的结果。但是，当我在一分钟后将1 bbb写入第一个流时，我什么都看不到。

我能正确理解updateStateByKey()的作用吗？或者我错了？

Answer 1

updateStateByKey完全符合您的要求。尤其是，如果当前窗口中不包含任何数据（strings.isEmpty()）你指示它忘记（返回Optional.absent();）：

if (!strings.isEmpty()) { 
    return Optional.of(strings.get(0)); 
} 
return Optional.absent(); 

，而你可能想的是回到以前的状态：

if (!strings.isEmpty()) { 
    return Optional.of(strings.get(0)); 
} 
return stringOptional;