On Thu, Oct 27, 2016 at 4:37 PM, Fabian Hueske <fhue...@gmail.com> wrote:
> Hi Luis, > > these blogposts should help you with the periodic partial result trigger > [1] [2]. > Hi, thanks for the links, I read them and tried to implement what I need, everything seems to work as expected except for the fact that the partial results aren't emitted, I created a gist with my PartialWindowTrigger implementation and the relevant part of the job: https://gist.github.com/anonymous/041987821e37ee8f862ce1857bb074ea Is the problem in the trigger? do I have to create a window assigner too? is it because of the windowAll? I reproduce part of the readme from the gist here for convenience, please see the readme for the PartialWindowTrigger implementation and the rest of the logs: What the job does (or I think it does) is to: - KeyBy the first string of a field - Create a tumbling window of 10 seconds - Register my PartialWindowTrigger that will trigger every 2 seconds (FIRE) and after the 10 second window (FIRE_AND_PURGE) - Fold on each partition to create a partial accumulation - Join all the partial results into a unique place through windowAll - Aggregate the partial aggregations into one result Here is the job's relevant part: return input.keyBy(keySelector) .timeWindow(Time.of(windowTime, timeUnit)) .trigger(new PartialWindowTrigger<>(partialWindowTime, timeUnit, windowTime, timeUnit)) .apply(creator.create(), timeWindowFold, timeWindowMerge) .windowAll(TumblingEventTimeWindows.of(Time.of(windowTime, timeUnit))) //.trigger(new PartialWindowTrigger<>(partialWindowTime, timeUnit, windowTime, timeUnit)) .apply(creator.create(), windowAllFold, windowAllMerge); The problem is that the triggers FIRE correctly but no partial results (every 2 seconds) are emitted, only the final result (every 10 seconds) is emitted. Even if instead of returning FIRE on onElement, I do: ctx.registerEventTimeTimer(timestamp); and return FIRE or FIRE_AND_PURGE on onEvent it still doesn't emit the partial values. There's a commented line on the job that registers a PartialWindowTrigger for the windowAll window but still doesn't work if uncommented. I added println's on the trigger and on the job steps, this is the output: 2016-11-03T11:07:04.180+01:00 onElement FIRE 2016-11-03T11:07:04.232+01:00 multiCountWindowFn 1 2016-11-03T11:07:04.305+01:00 windowAllFold 2016-11-03T11:07:04.733+01:00 timeWindowFold 2016-11-03T11:07:04.681+01:00 onElement CONTINUE 2016-11-03T11:07:05.234+01:00 timeWindowFold 2016-11-03T11:07:05.182+01:00 onElement CONTINUE 2016-11-03T11:07:05.735+01:00 timeWindowFold 2016-11-03T11:07:05.682+01:00 onElement CONTINUE 2016-11-03T11:07:06.236+01:00 timeWindowFold 2016-11-03T11:07:06.183+01:00 onElement FIRE <3 more blocks like the one above here> 2016-11-03T11:07:12.246+01:00 multiCountWindowFn 1 2016-11-03T11:07:12.317+01:00 windowAllFold 2016-11-03T11:07:12.746+01:00 timeWindowFold 2016-11-03T11:07:12.693+01:00 onElement CONTINUE 2016-11-03T11:07:13.247+01:00 timeWindowFold 2016-11-03T11:07:13.194+01:00 onElement CONTINUE 2016-11-03T11:07:13.748+01:00 timeWindowFold 2016-11-03T11:07:13.695+01:00 onElement CONTINUE 2016-11-03T11:07:09.999+01:00 onEventTime FIRE_AND_PURGE 2016-11-03T11:07:13.948+01:00 multiCountWindowFn 1 2016-11-03T11:07:14.020+01:00 windowAllFold 2016-11-03T11:07:14.020+01:00 allWindowMerger 1 {"blue":{"foo":{"v":65}},"$":{"ts":1478167634020}} > Regarding the second question: > Time windows are by default aligned to 1970-01-01-00:00:00. > So a 24 hour window will always start at 00:00. > > Best, Fabian > > [1] http://flink.apache.org/news/2015/12/04/Introducing-windows.html > [2] https://www.mapr.com/blog/essential-guide-streaming- > first-processing-apache-flink > > 2016-10-27 16:31 GMT+02:00 Luis Mariano Guerra <mari...@event-fabric.com>: > >> hi, >> >> I need to calculate some counts for the day but also emit the partial >> counts periodically, I think triggers may help me, I'm searching around but >> there's not much content about it, any tip? >> >> for example I'm counting access by location to different services, I want >> to accumulate access during the whole day, but I want to emit the partial >> count every 5 minutes. >> >> one slightly related question, is there a way to align a window to a day? >> for example a 24 hour window that starts at 00:00. >> >> thanks. >> > >
