Wie führen wir benutzerdefinierte Operationen auf WindowedStream in Flink durch?

Question

Ich möchte einige Operationen auf einem WindowedStream in Flink, sagen Durchschnitt. Aber es gibt sehr begrenzte Operationen zur Verfügung vordefiniert wie Summe, min, max usw.

val windowedStream = valueStream 
          .keyBy(0) 
          .timeWindow(Time.minutes(5)) 
          .sum(2) //Change this to average? 

Angenommen, ich möchte den Durchschnitt finden, wie kann ich das tun?

Answer 1

Flink hat keine eingebaute Funktion, um den Durchschnitt auf einer WindowStream zu berechnen. Dazu müssen Sie eine benutzerdefinierte WindowFunction implementieren.

Der effizienteste Weg ist, eine ReduceFunction zu implementieren, die die Zählung und Summe aus dem Wert berechnen, den Sie im Durchschnitt und einem nachfolgenden WindowFunction möchte, dass das Ergebnis der ReduceFunction und berechnet den Durchschnitt dauert. Die Verwendung eines ReduceFunction ist effizienter, da Flink es direkt auf eingehende Werte anwendet. Daher aggregiert es Werte im laufenden Betrieb und sammelt sie nicht im Fenster. Dies reduziert den Speicherbedarf eines Fensters erheblich.

Da die Ausgabe eines ReduceFunction denselben Typ wie seine Eingabe hat, müssen Sie ein Feld für die Zählung hinzufügen, bevor Sie die ReduceFunction anwenden.

etwas wie die folgenden sollte es tun:

val valueStream: DataStream[(String, Double)] = ??? 

val r: DataStream[(String, Double)] = valueStream 
    // append a 1L for counting 
    .map(x => (x._1, x._2, 1l)) 
    // key and window stream 
    .keyBy(0).timeWindow(Time.minutes(5)) 
    .apply(
    // ReduceFunction (compute sum and count) 
    (x: (String, Double, Long), y: (String, Double, Long)) => 
     (x._1, x._2 + y._2, x._3 + y._3), 
    // WindowFunction 
    (key, window: TimeWindow, input: Iterable[(String, Double, Long)], out: Collector[(String, Double)]) => { 
     // get first (and only) value 
     val x: (String, Double, Long) = input.toIterator.next 
     // compute average as sum/count 
     out.collect(x._1, x._2/x._3) 
    } 
)