Vergleiche neuen Wert mit dem vorherigen Durchschnitt in Spark Streaming

Question

Ich habe ein Fenster in Spark Streaming mit 5 ganzzahligen Werten erstellt. Jedes Mal, wenn eine Sekunde verstrichen ist, erhält das Fenster einen neuen Wert und verliert den ältesten. Jedes Mal, wenn ein neuer Wert kommt, berechne ich den Durchschnitt der Werte im Fenster wie folgt aus:

JavaDStream<Integer> average = values.map(new Function<Integer, Integer>() { 
     @Override 
     public Integer call(Integer a) throws Exception { 
      int b = a/5; 


      return b; 


    } }); 

So ist die durchschnittliche Veränderung hält.

Das Problem ist, jedes Mal, wenn ein neuer Wert zum Fenster kommt, möchte ich es mit dem vorherigen Durchschnitt vergleichen. Wenn dieser neue Wert viel größer oder kleiner als der Durchschnittswert ist, möchte ich ihn fallen lassen und den alten Durchschnitt beibehalten. Wenn nicht, kann die Durchschnitt aktualisiert werden.

Meine Frage ist, wie kann ich diesen 'alten Durchschnitt' speichern, damit ich ihn mit dem neuen Wert im Fenster vergleichen kann?

Vielen Dank.

Answer 1

Sie etwas versuchen könnten, ähnlich den folgenden:

1. einen „Schlüssel-Wert DSTREAM“ erstellen mit einer Pseudo-Taste (zB 1)
2. Verwendung mapWithState() den letzten Durchschnitt zu speichern und zu vergleichen, wie du erwähntest; Rückkehr der Durchschnitt Sie

es wie folgt aussehen könnte weiter verwenden möchten:

JavaPairDStream<Integer, Integer> pairedIntStream = values.mapToPair(
new PairFunction<Integer, Integer, Integer>() { 
     @Override 
     public Tuple2<Integer, Integer> call(Integer a) throws Exception { 
      return new Tuple2<Integer, Integer>(1, a); 
} }); 

Function3<Integer, Optional<Integer>, State<Integer>, Integer> mappingFunction = 
     new Function3<Integer, Optional<Integer>, State<Integer>, Integer>() { 
      @Override 
      public Integer call(Integer s, Optional<Integer> value, State<Integer> state) { 

       int avg = value.orNull()/5; 

       if(!state.exists() || state.get() < avg) { 
        state.update(avg); 
        return avg; 
       } 
       return state.get(); 
      } 
     }; 

    JavaMapWithStateDStream<Integer, Integer, Integer, Integer> mapWithStateDStream = 
     pairedIntStream.mapWithState(StateSpec.function(mappingFunc));