Hi Team, My team has a use case where we will have a DataStream of Numbers and we want to do the aggregation on it and return the aggregated value. I have tried different approaches using reduce functions, aggregate functions and reducingState. Below I have discussed it in detail

Case 1: In Reduced function the output is dataStream of rolling Sum. To get the final Sum I have to traverse the outputStream and the last value would be my total . In my case, I don't want to iterate the whole dataStream to get the final Sum and also don't wan't to use an extra DataStream to just store the final aggregated value.

Case 2: I am able to access aggregate() method only after countWindow() and in countWindow() we have to pass the size. As I don't know the size of my dataStream(User will be sending data to me) I can't use it.

Below is my implementation of reduce function

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStreamSource<Double> dataStream = env.fromElements(2.00, 3.00, 4.00, 11.00, 13.00, 14.00);

DataStream<Double> singleOutputStreamOperator = dataStream.keyBy(value -> "key").reduce(new ReduceFunction<Double>() {
  @Override
  public Double reduce(Double aDouble, Double t1) throws Exception {
    return aDouble+ t1;
  }
});

singleOutputStreamOperator.print();
DataStream.Collector<Double> doubleCollector = new DataStream.Collector<>();
singleOutputStreamOperator.collectAsync(doubleCollector);
singleOutputStreamOperator.executeAndCollect("Aggregation");

Double result = null;
while( doubleCollector.getOutput().hasNext() ) {
  result = doubleCollector.getOutput().next();
  System.out.println("result = " + result);
}

In above implementation result would be my final aggregation value. The output looks like : 2.0, 5.0, 9.0, 20.0, 33.0. I simply want to get 33.0 as my aggregated value and store it in variable and give it to user. Is there any way I can implement it

You simultaneously asked this question here in Slack in both the troubleshooting channel and in the dev channel (where it does not belong), as well as on Stack Overflow (where I answered it), and on the user mailing list (where Liu Ron answered it). Please keep in mind that questions in these forums are answered on a best effort basis by volunteers. Asking the same question in multiple forums causes duplicated effort within the community. In the future, kindly limit yourself to one of these forums. Commercial support for Flink is available elsewhere, should you require more timely responses.

Hi David, Apologies for the inconvenience caused. I will keep it in mind going forward. Thanks again