Last-Event-Only Timer (Custom Trigger)

[Julian Bauß]

Hello everybody,

I'm currently trying to implement a Function that allows me to detect that a certain amount of time has passed after receiving the last element of a stream (in a given time window). For example if nothing happened for 6 hours within a given Session I want to do something (set a flag, clear some state).

I thought I could solve this with a custom trigger on EventTime TimeWindows. I'm currently confused about how I should implement that Trigger. The implementation should not be much different from a EventTimeTrigger except that it discards of any windows with more than one element. This would lead to a windowing mechanism that effectively only fires a window after a certain time for the last element.

What I don't understand is when the regular EventTimeTrigger purges windows because it only ever returns FIRE and CONTINUE events.

I assumed that after firing a window onEventTime the window would get purged eventually. I then would've added a PURGE once the number of elements was greater than 1.

Would that be a suitable implementation?

Best Regards,

Julian

[Till Rohrmann]

Hi Julian,

you can use the TriggerContext to register and unregister event time timers which fire when the given event time has been passed. That’s one way to implement what you’ve described. If you don’t want to use time windows you could also use session windows. Take a look at the EventTimeSessionWindows class. In order to only obtain the last element, you should use an Evictor which evicts all elements except for the last.

Concerning the purging: Time windows are automatically cleaned up after the end of the window + an allowed lateness. That’s why the trigger no longer has to take core of that.

Cheers,
Till

​

On Mon, Nov 7, 2016 at 11:34 AM, Julian Bauß <[hidden email]> wrote:

Hello everybody,

I'm currently trying to implement a Function that allows me to detect that a certain amount of time has passed after receiving the last element of a stream (in a given time window). For example if nothing happened for 6 hours within a given Session I want to do something (set a flag, clear some state).

I thought I could solve this with a custom trigger on EventTime TimeWindows. I'm currently confused about how I should implement that Trigger. The implementation should not be much different from a EventTimeTrigger except that it discards of any windows with more than one element. This would lead to a windowing mechanism that effectively only fires a window after a certain time for the last element.

What I don't understand is when the regular EventTimeTrigger purges windows because it only ever returns FIRE and CONTINUE events.

I assumed that after firing a window onEventTime the window would get purged eventually. I then would've added a PURGE once the number of elements was greater than 1.

Would that be a suitable implementation?

Best Regards,

Julian

[Julian Bauß]

Hi Till,

thank you for your reply.

This is exactly what I was looking for!

Flink continues to surprise me with its versatility. :)

Best Regards,

Julian

2016-11-07 16:47 GMT+01:00 Till Rohrmann <[hidden email]>:

Hi Julian,

you can use the TriggerContext to register and unregister event time timers which fire when the given event time has been passed. That’s one way to implement what you’ve described. If you don’t want to use time windows you could also use session windows. Take a look at the EventTimeSessionWindows class. In order to only obtain the last element, you should use an Evictor which evicts all elements except for the last.

Concerning the purging: Time windows are automatically cleaned up after the end of the window + an allowed lateness. That’s why the trigger no longer has to take core of that.

Cheers,
Till

​

On Mon, Nov 7, 2016 at 11:34 AM, Julian Bauß <[hidden email]> wrote:

Hello everybody,

I'm currently trying to implement a Function that allows me to detect that a certain amount of time has passed after receiving the last element of a stream (in a given time window). For example if nothing happened for 6 hours within a given Session I want to do something (set a flag, clear some state).

I thought I could solve this with a custom trigger on EventTime TimeWindows. I'm currently confused about how I should implement that Trigger. The implementation should not be much different from a EventTimeTrigger except that it discards of any windows with more than one element. This would lead to a windowing mechanism that effectively only fires a window after a certain time for the last element.

What I don't understand is when the regular EventTimeTrigger purges windows because it only ever returns FIRE and CONTINUE events.

I assumed that after firing a window onEventTime the window would get purged eventually. I then would've added a PURGE once the number of elements was greater than 1.

Would that be a suitable implementation?

Best Regards,

Julian