Understanding code of CountTrigger

[nsengupta]

Hello all,

Here's a code comment from org.apache.flink.streaming.api.windowing.triggers.Trigger:

/**

         * Result type for trigger methods. This determines what happens which the window.

         *

         * <p>

         * On {@code FIRE} the pane is evaluated and results are emitted. The contents of the window

         * are kept. {@code FIRE_AND_PURGE} acts like {@code FIRE} but the contents of the pane

         * are purged. On {@code CONTINUE} nothing happens, processing continues. On {@code PURGE}

         * the contents of the window are discarded and now result is emitted for the window.

*/

And, here's the code snippet from org.apache.flink.streaming.api.windowing.triggers.CountTrigger:

@Override

        public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws IOException {

                OperatorState<Long> count = ctx.getKeyValueState("count", 0L);

                long currentCount = count.value() + 1;

                count.update(currentCount);

                if (currentCount >= maxCount) {

                        count.update(0L);

                        return TriggerResult.FIRE;

                }

                return TriggerResult.CONTINUE;

        }

Following the code-comment, I understand that elements are not PURGEd from a CountWindow, only FIREd. The contents on the Window stay forever.

Now, that is counter-intuitive to me. Something is not right about this.

What am I missing? Help me to plug the holes in my understanding.

-- Nirmalya

--

[Till Rohrmann]

Hi Nirmalya,

the CountTrigger always works together with the CountEvictor which will make sure that only count elements are kept in the window. Evictors can evict elements from the window after the trigger event. That is the reason why the CountTrigger does not have to purge the window explicitly.

Cheers,
Till

​

On Wed, Feb 3, 2016 at 2:36 AM, Nirmalya Sengupta <[hidden email]> wrote:

Hello all,

Here's a code comment from org.apache.flink.streaming.api.windowing.triggers.Trigger:

/**

         * Result type for trigger methods. This determines what happens which the window.

         *

         * <p>

         * On {@code FIRE} the pane is evaluated and results are emitted. The contents of the window

         * are kept. {@code FIRE_AND_PURGE} acts like {@code FIRE} but the contents of the pane

         * are purged. On {@code CONTINUE} nothing happens, processing continues. On {@code PURGE}

         * the contents of the window are discarded and now result is emitted for the window.

*/

And, here's the code snippet from org.apache.flink.streaming.api.windowing.triggers.CountTrigger:

@Override

        public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws IOException {

                OperatorState<Long> count = ctx.getKeyValueState("count", 0L);

                long currentCount = count.value() + 1;

                count.update(currentCount);

                if (currentCount >= maxCount) {

                        count.update(0L);

                        return TriggerResult.FIRE;

                }

                return TriggerResult.CONTINUE;

        }

Following the code-comment, I understand that elements are not PURGEd from a CountWindow, only FIREd. The contents on the Window stay forever.

Now, that is counter-intuitive to me. Something is not right about this.

What am I missing? Help me to plug the holes in my understanding.

-- Nirmalya

--

Software Technologist
http://www.linkedin.com/in/nirmalyasengupta
"If you have built castles in the air, your work need not be lost. That is where they should be.
Now put the foundation under them."

[nsengupta]

Hello Till <[hidden email]>,

From your prompt reply:

'... the CountTrigger always works together with the CountEvictor which will make sure that only .. ' - that explains it. Thanks.  I missed it.

A related question I have is this:

Between the PURGE facility of Trigger and REMOVAL facility of Evictor, is there really a difference? Who uses the value that an 

Evictor.evict(Iterable<StreamRecord<Object>> elements, int size, W window) {

.....

}

returns? A trigger doesn't seem to do it.

-- Nirmalya

--

[Aljoscha Krettek]

Hi Nirmalya,
the result of Evictor.evict() is used internally by the window operator. The flow is as follows:
1. Trigger fires
2. Evictor is called if it exists
3. Elements are evicted from window buffer if evictor returned number > 0
4. User-provided window function is called to emit window results

Cheers,
Aljoscha

> On 03 Feb 2016, at 12:46, Nirmalya Sengupta <[hidden email]> wrote:
>
> Hello Till <[hidden email]>,
>
> From your prompt reply:
>
> '... the CountTrigger always works together with the CountEvictor which will make sure that only .. ' - that explains it. Thanks.  I missed it.
>
> A related question I have is this:
>
> Between the PURGE facility of Trigger and REMOVAL facility of Evictor, is there really a difference? Who uses the value that an
>
> Evictor.evict(Iterable<StreamRecord<Object>> elements, int size, W window) {
> .....
>
> }
>
> returns? A trigger doesn't seem to do it.
>
> -- Nirmalya
>
>
>
>
>
>
>
> --
> Software Technologist
> http://www.linkedin.com/in/nirmalyasengupta
> "If you have built castles in the air, your work need not be lost. That is where they should be.
> Now put the foundation under them."

[nsengupta]

Hello Aljoscha <[hidden email]>,

Many thanks for the explanation. Referring to the flow from your response:

---------------------------------------------------------------

1. Trigger fires
2. Evictor is called if it exists
3. Elements are evicted from window buffer if evictor returned number > 0
4. User-provided window function is called to emit window results

-----------------------------------------------------------------

Just one clarification:

If my Trigger acts as FIRE_AND_PURGE, when exactly does the PURGing happen: between <1> and <2> above or between <4> and <1> above? My understanding is that it should happen between <4> and <1> because User-provided window function must be able to see the data to process it, before window buffer is emptied.

-- Nirmalya

--

[nsengupta]

Hello Aljoscha <[hidden email]>,

I have checked again with the (fantastic) blog here: https://flink.apache.org/news/2015/12/04/Introducing-windows.html

and I have come to understand that the contents of a window-buffer must be disposed of only after the User-defined evaluation function has 

seen and used them all; in other words, referring to the flow in my mail below, it happens between step <4> and <1>.

Spurred by that sudden 'Aha' moment, I thought I should write to you again, so that you have to take less effort to explain. :-)

-- Nirmalya

<*********  my earlier mail on the same thread ***********>

Hello Aljoscha <[hidden email]>,

Many thanks for the explanation. Referring to the flow from your response:

---------------------------------------------------------------

1. Trigger fires
2. Evictor is called if it exists
3. Elements are evicted from window buffer if evictor returned number > 0
4. User-provided window function is called to emit window results

-----------------------------------------------------------------

Just one clarification:

If my Trigger acts as FIRE_AND_PURGE, when exactly does the PURGing happen: between <1> and <2> above or between <4> and <1> above? My understanding is that it should happen between <4> and <1> because User-provided window function must be able to see the data to process it, before window buffer is emptied.

-- Nirmalya

--

Software Technologist
http://www.linkedin.com/in/nirmalyasengupta
"If you have built castles in the air, your work need not be lost. That is where they should be.
Now put the foundation under them."

--

[Aljoscha Krettek]

Very good, you are absolutely right. :D
> On 04 Feb 2016, at 05:07, Nirmalya Sengupta <[hidden email]> wrote:
>
> on