Allowed Lateness and Window State

[swiesman]

Hi all,

I have a question about allowed lateness semantics and window state. I have a custom event time trigger that does an early FIRE onElement for a certain set of conditions and then a FIRE_AND_PURGE onEventTime.

  stream
    .keyBy(???)
    .timeWindow(Time.hour(1))
    .trigger(new Trigger {
        def onElement(element: Int, timestamp: Long, window: TimeWindow, ctx: TriggerContext): TriggerResult {
            ctx.registerEventTimeTimer(window.getEnd)
            
            if (/*some condition*/){
                TriggerResult.FIRE
            } else {
                TriggerResult.CONTINUE
            }
        }

        def def onEventTime(time: Long, window: TimeWindow, ctx: TriggerContext): TriggerResult =
            TriggerResult.FIRE_AND_PURGE
    })

And this has worked great for me. However I now want to add a certain amount of allowed lateness to this window. My understanding has been that when TriggerResult.FIRE_AND_PURGE is returned the window will be fully evaluated and the state cleared. So if a new element shows up in the allowed lateness time will it be put into a new singleton window and evaluated or does specifying allowed lateness defer the window purge until after the allowed lateness has passed?

Put another way, in the illustration below if my window is counting circles after the red circle is seen will my window output 5 or 1? If I want it to output 5 do I have to modify my trigger?


Thank you,

Seth Wiesman

[Kostas Kloudas]

Hi Seth,

FIRE and FIRE_AND_PURGE still have the same meaning.
So on eventTime, when your trigger says FIRE_AND_PURGE, your window will be
evaluated (4) and its state will be purged. Now, when the red circle arrives, the state
will be empty, so if the onElement says FIRE, the result will be 1.

If you want to keep also the previous elements and have
4 on the onEventTime, and 5 onElement for the red circle, then you just
have to modify your trigger to return FIRE onEventTime, instead of FIRE_AND_PURGE.

When the allowed lateness expires, your window will be garbage collected
(all its state will be purged). So your trigger does not need to worry about that.

I hope this answers your question.

Cheers,
Kostas

> On Oct 13, 2016, at 4:16 PM, swiesman <[hidden email]> wrote:
>
> Hi all,
>
> I have a question about allowed lateness semantics and window state. I have
> a custom event time trigger that does an early FIRE onElement for a certain
> set of conditions and then a FIRE_AND_PURGE onEventTime.
>
>
>
>
> And this has worked great for me. However I now want to add a certain amount
> of allowed lateness to this window. My understanding has been that when
> TriggerResult.FIRE_AND_PURGE is returned the window will be fully evaluated
> and the state cleared. So if a new element shows up in the allowed lateness
> time will it be put into a new singleton window and evaluated or does
> specifying allowed lateness defer the window purge until after the allowed
> lateness has passed?
>
> Put another way, in the illustration below if my window is counting circles
> after the red circle is seen will my window output 5 or 1? If I want it to
> output 5 do I have to modify my trigger?
>
>
> Thank you,
>
> Seth Wiesman
> <http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/file/n9529/14763682863543.png>
>
>
>
> --
> View this message in context: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Allowed-Lateness-and-Window-State-tp9529.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive at Nabble.com.

[swiesman]

Thank makes perfect sense, thank you so much for your quick response Kostas!

Seth Wiesman