[ 
https://issues.apache.org/jira/browse/FLINK-4101?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15346204#comment-15346204
 ] 

Vasia Kalavri commented on FLINK-4101:
--------------------------------------

Hi [~mrakki3110], JIRA is used for reporting bugs or proposing new features. 
Your question should be posted either in the user mailing list or SO, as you've 
already done. I'm closing this.

> 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", "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 calcluate average temperature for the above example ??



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Reply via email to