Hey Aljoscha, the first solution did not work out as expected. As when late elements arrive the first window is triggered again and would emit a new (accumulated) event, that would be counted twice (in time accumulation and late accumulation) in the second window.I could implement my own (discarding strategy) like in Apache Beam, but the out stream should contain accumulated events that are stored in cassandra. The second solution just gave an compiler error, thus I think is not possible right now.
best Stephan > On 21 Nov 2016, at 17:56, Aljoscha Krettek <[email protected]> wrote: > > Hi, > why did you settle for the last solution? > > Cheers, > Aljoscha > > On Thu, 17 Nov 2016 at 15:57 kaelumania <[email protected] > <mailto:[email protected]>> wrote: > Hi Fabian, > > your proposed solution for: > > Multiple window aggregations > You can construct a data flow of cascading window operators and fork off (to > emit or further processing) the result after each window. > > Input -> window(15 secs) -> window(1 min) -> window(15 min) -> ... > \-> out_1 \-> out_2 \-> out_3 > does not work, am I missing something? > > First I tried the following > DataStream<Reading> values = input.assignTimestampsAndWatermarks(new > StrictWatermarkAssigner()); // force lateness > > DataStream<ReadingAggregate> aggregatesPerMinute = values > .keyBy("id") > .timeWindow(Time.minutes(1)) > .allowedLateness(Time.minutes(2)) > .apply(new ReadingAggregate(), new AggregateReadings(), new > AggregateReadings()); > > DataStream<ReadingAggregate> aggregatesPerHour = aggregatesPerMinute > .keyBy("id") > .timeWindow(Time.hours(1)) > .allowedLateness(Time.hours(2)) > .apply(new AggregateReadingAggregates(), new > AggregateReadingAggregates()); > but due to late data the first fold function would emit 2 rolling aggregates > (one with and one without the late element), which results in being counted > twice within the second reducer. Therefore i tried > WindowedStream<Reading, Tuple, TimeWindow> readingsPerMinute = input > .assignTimestampsAndWatermarks(new StrictWatermarkAssigner()) // > force lateness > .keyBy("id") > .timeWindow(Time.minutes(1)) > .allowedLateness(Time.hours(2)); > > WindowedStream<Reading, Tuple, TimeWindow> readingsPerHours = > readingsPerMinute > .timeWindow(Time.hours(1)) > .allowedLateness(Time.hours(2)); > > DataStream<ReadingAggregate> aggregatesPerMinute = > readingsPerMinute.apply(new ReadingAggregate(), new AggregateReadings(), new > AggregateReadings()); > DataStream<ReadingAggregate> aggregatesPerHour = readingsPerHours.apply(new > ReadingAggregate(), new AggregateReadings(), new AggregateReadings()); > which gives me a compiler error as WindowedStream does not provide a > timeWindow method. > > Finally I settled with this: > KeyedStream<Reading, Tuple> readings = input > .assignTimestampsAndWatermarks(new StrictWatermarkAssigner()) // > force lateness > .keyBy("id"); > > DataStream<ReadingAggregate> aggregatesPerMinute = readings > .timeWindow(Time.minutes(1)) > .allowedLateness(Time.hours(2)) > .apply(new ReadingAggregate(), new AggregateReadings(), new > AggregateReadings()); > > DataStream<ReadingAggregate> aggregatesPerHour = readings > .timeWindow(Time.hours(1)) > .allowedLateness(Time.hours(2)) > .apply(new ReadingAggregate(), new AggregateReadings(), new > AggregateReadings()); > > > Feedback is very welcome. > > best, Stephan > > > > >> On 11 Nov 2016, at 00:29, Fabian Hueske-2 [via Apache Flink User Mailing >> List archive.] <[hidden email] >> <http://user/SendEmail.jtp?type=node&node=10179&i=0>> wrote: >> > >> Hi Stephan, >> >> I just wrote an answer to your SO question. >> >> Best, Fabian > >> >> 2016-11-10 11:01 GMT+01:00 Stephan Epping <<a >> href="x-msg://3/user/SendEmail.jtp?type=node&node=10033&i=0" >> target="_top" rel="nofollow" link="external" class="">[hidden email]>: > >> >> Hello, >> >> I found this question in the Nabble archive >> (http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Maintaining-watermarks-per-key-instead-of-per-operator-instance-tp7288.html >> >> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Maintaining-watermarks-per-key-instead-of-per-operator-instance-tp7288.html>) >> but was unable/dont know how to reply. >> >> Here is my question regarding the mentioned thread: >> >>> Hello, >>> >>> I have similar requirements (see StackOverflor >>> http://stackoverflow.com/questions/40465335/apache-flink-multiple-window-aggregations-and-late-data >>> >>> <http://stackoverflow.com/questions/40465335/apache-flink-multiple-window-aggregations-and-late-data>). >>> I am pretty new to flink, could you elaborate on a possible solution? We >>> can guarantee good ordering by sensor_id, thus watermarking by key would be >>> the only reasonable way for us >>> (sensorData.keyBy('id').timeWindow(1.minute).sum('value')), could I do my >>> own watermarking aftersensorData.keyBy('id').overwriteWatermarking()... per >>> key? Or maybe using custom state plus a custom trigger? What happens if a >>> sensor dies or is being removed completely, how can this be detected as >>> watermarks would be ignored for window garbage collection. Or could we >>> dynamically schedule a job of each sensor? Which would result in 1000 Jobs. >> >> >> Thanks, >> Stephan >> >> > >> If you reply to this email, your message will be added to the discussion >> below: > >> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Maintaining-watermarks-per-key-instead-of-per-operator-instance-tp7288p10033.html >> >> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Maintaining-watermarks-per-key-instead-of-per-operator-instance-tp7288p10033.html> >> To unsubscribe from Maintaining watermarks per key, instead of per operator >> instance, click here <>. >> NAML >> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/template/NamlServlet.jtp?macro=macro_viewer&id=instant_html%21nabble%3Aemail.naml&base=nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.naml.namespaces.BasicNamespace-nabble.view.web.template.NabbleNamespace-nabble.view.web.template.NodeNamespace&breadcrumbs=notify_subscribers%21nabble%3Aemail.naml-instant_emails%21nabble%3Aemail.naml-send_instant_email%21nabble%3Aemail.naml> > View this message in context: Re: Maintaining watermarks per key, instead of > per operator instance > <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Maintaining-watermarks-per-key-instead-of-per-operator-instance-tp7288p10179.html> > Sent from the Apache Flink User Mailing List archive. mailing list archive > <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/> at > Nabble.com.
