Apache Spark-merge nach updateStateByKey()

Question

Ich versuche, zwei Ströme zu verschmelzen und einer von ihnen sollte Stateful sein (wie statische Daten nicht häufige Aktualisierungen):

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(); 

Wenn ich schreibe 1 aaa in den ersten Strom und 1 111 in der Sekunde sofort alles funktioniert gut, sehe ich Ergebnis der Zusammenführung. Aber wenn ich 1 bbb in den ersten Stream nach einer Minute schreibe, sehe ich nichts.

Verstehe ich richtig, was updateStateByKey() tut? Oder ich liege falsch?

Answer 1

updateStateByKey tut genau das, wonach Sie es verlangen. Insbesondere dann, wenn aktuelle Fenster keine Daten enthält (strings.isEmpty()) Sie es (Return Optional.absent();) zu vergessen anweisen:

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

während das, was Sie wahrscheinlich wollen vorherigen Zustand zurückzukehren:

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