Calculate average using Flink DataStream for window duration

Question

I use the Flink DataStream API where racks are available, and I want to calculate the “average” from the temperature group by rack identifiers. My window time is 40 seconds, and my window expands every 10 seconds ... Below is my code where I calculate the sum of the temperatures every 10 seconds for each rack, but now I want to calculate the <strong> average temperatures:

static Properties properties=new Properties();
    public static Properties getProperties()
    {
        properties.setProperty("bootstrap.servers", "54.164.200.104:9092");
        properties.setProperty("zookeeper.connect", "54.164.200.104:2181");
        //properties.setProperty("deserializer.class", "kafka.serializer.StringEncoder");
        //properties.setProperty("group.id", "akshay");
        properties.setProperty("auto.offset.reset", "earliest");
        return properties;
    }

 @SuppressWarnings("rawtypes")
public static void main(String[] args) throws Exception 
{
    StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    Properties props=Program.getProperties();
    DataStream<TemperatureEvent> dstream=env.addSource(new FlinkKafkaConsumer09<TemperatureEvent>("TemperatureEvent", new TemperatureEventSchema(), props)).assignTimestampsAndWatermarks(new IngestionTimeExtractor<>());
    DataStream<TemperatureEvent> ds1=dstream.keyBy("rackId").timeWindow(Time.seconds(40), Time.seconds(10)).sum("temperature");
    env.execute("Temperature Consumer");
}

How can I calculate the average temperature for the above example?

Answer 1

, . :

https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/GroupedProcessingTimeWindowExample.java

, , .sum("temperature");

- .apply(new Avg()); Avg:

public class Avg implements WindowFunction<TemperatureEvent,  TemperatureEvent, Long, org.apache.flink.streaming.api.windowing.windows.Window> {

  @Override
  public void apply(Long key, Window window, Iterable<TemperatureEvent> values, Collector<TemperatureEvent> out) {
    long sum = 0L;
    int count = 0;
    for (TemperatureEvent value : values) {
        sum += value.getTemperature();
        count ++;
    }

    TemperatureEvent result = values.iterator().next();
    result.setTemperature(sum / count);
    out.collect(result);
  }
}

: , (, ), .head