我不确定我必须使用哪个stream Flink transformation来计算某个流的平均值并在5秒的时间内更新一个状态(假设它是一个int array)。

streamSource

.keyBy(0)

.flatMap(new MyRichFlatMapFunction())

.print()

我不能把它写在窗户上。

streamSource

.keyBy(0)

.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

.aggregate(new MyAggregateFunction())

.print()

我无法通过ValueState保持 array状态。

我试图使用RichAggregateFunction，但遇到了这个线程的相同问题。使用RichAggregateFunction时的Flink错误有没有其他方法来计算平均值并跟踪Flink中的另一个状态？

在 Flink我该如何处理这个问题？

streamStations.filter(new SensorFilter("COUNT_TR"))

.map(new TrainStationMapper())

.keyBy(new MyKeySelector())

.window(TumblingEventTimeWindows.of(Time.seconds(5)));

// THIS AGGREGATE DOES NOT WORK

// .aggregate(new AverageRichAggregator())

// .print();

public static class AverageRichAggregator extends

RichAggregateFunction<Tuple3<Integer, Tuple5<Integer, String, Integer, String, Integer>, Double>, Tuple3<Double, Long, Integer>, Tuple2<String, Double>> {

private static final long serialVersionUID = -40874489412082797L;

private String functionName;

private ValueState<CountMinSketch> countMinSketchState;

@Override

public void open(Configuration parameters) throws Exception {

ValueStateDescriptor<CountMinSketch> descriptor = new ValueStateDescriptor<>("countMinSketchState",

CountMinSketch.class);

this.countMinSketchState = getRuntimeContext().getState(descriptor);

}

@Override

public Tuple3<Double, Long, Integer> createAccumulator() {

this.countMinSketchState.clear();

return new Tuple3<>(0.0, 0L, 0);

}

@Override

public Tuple3<Double, Long, Integer> add(

Tuple3<Integer, Tuple5<Integer, String, Integer, String, Integer>, Double> value,

Tuple3<Double, Long, Integer> accumulator) {

try {

if (value.f1.f1.equals("COUNT_PE")) {

// int count = (int) Math.round(value.f2);

// countMinSketch.updateSketchAsync("COUNT_PE");

} else if (value.f1.f1.equals("COUNT_TI")) {

// int count = (int) Math.round(value.f2);

// countMinSketch.updateSketchAsync("COUNT_TI");

} else if (value.f1.f1.equals("COUNT_TR")) {

// int count = (int) Math.round(value.f2);

// countMinSketch.updateSketchAsync("COUNT_TR");

}

CountMinSketch currentCountMinSketchState = this.countMinSketchState.value();

currentCountMinSketchState.updateSketchAsync(value.f1.f1);

this.countMinSketchState.update(currentCountMinSketchState);

} catch (IOException e) {

e.printStackTrace();

}

return new Tuple3<>(accumulator.f0 + value.f2, accumulator.f1 + 1L, value.f1.f4);

}

@Override

public Tuple2<String, Double> getResult(Tuple3<Double, Long, Integer> accumulator) {

String label = "";

int frequency = 0;

try {

if (functionName.equals("COUNT_PE")) {

label = "PEOPLE average on train station";

// frequency = countMinSketch.getFrequencyFromSketch("COUNT_PE");

} else if (functionName.equals("COUNT_TI")) {

label = "TICKETS average on train station";

// frequency = countMinSketch.getFrequencyFromSketch("COUNT_TI");

} else if (functionName.equals("COUNT_TR")) {

label = "TRAIN average on train station";

// frequency = countMinSketch.getFrequencyFromSketch("COUNT_TR");

}

frequency = this.countMinSketchState.value().getFrequencyFromSketch(functionName);

} catch (IOException e) {

e.printStackTrace();

}

return new Tuple2<>(label + "[" + accumulator.f2 + "] reads[" + frequency + "]",

((double) accumulator.f0) / accumulator.f1);

}

@Override

public Tuple3<Double, Long, Integer> merge(Tuple3<Double, Long, Integer> a, Tuple3<Double, Long, Integer> b) {

return new Tuple3<>(a.f0 + b.f0, a.f1 + b.f1, a.f2);

}

}

错误：

Exception in thread "main" java.lang.UnsupportedOperationException: This aggregation function cannot be a RichFunction.

at org.apache.flink.streaming.api.datastream.WindowedStream.aggregate(WindowedStream.java:692)

at org.sense.flink.examples.stream.MultiSensorMultiStationsReadingMqtt2.(MultiSensorMultiStationsReadingMqtt2.java:71)

at org.sense.flink.App.main(App.java:141)

谢谢