flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Akshay Shingote (JIRA)" <j...@apache.org>
Subject [jira] [Created] (FLINK-4101) Calculating average in Flink DataStream on window time
Date Wed, 22 Jun 2016 14:10:57 GMT
Akshay Shingote created FLINK-4101:

             Summary: Calculating average in Flink DataStream on window time
                 Key: FLINK-4101
                 URL: https://issues.apache.org/jira/browse/FLINK-4101
             Project: Flink
          Issue Type: Task
          Components: DataStream API
    Affects Versions: 1.0.2
            Reporter: Akshay Shingote

I am using Flink DataStream API where there where racks are available & I want to calculate
"average"of temperature group by rack IDs. My window duration is of 40 seconds & my window
is sliding every 10 seconds...Following is my code where I am calculating sum of temperatures
every 10 seconds for every rackID,but now I want to calculate average temperatures::

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

public static void main(String[] args) throws Exception 
    StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
    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),
    env.execute("Temperature Consumer");
How can I calcluate average temperature for the above example ??

This message was sent by Atlassian JIRA

View raw message