Operation of Windows and Triggers

[Piyush Shrivastava]

I wanted to know how Windows and Triggers work in Flink. I am creating a time window of 20 seconds and a count trigger of 100.

stream.keyBy(0)
             .timeWindow(Time.seconds(20))
             .trigger(CountTrigger.of(100))

In this case, when will my window get triggered? When 20 seconds has passed, 100 messages are passed?

[Fabian Hueske-2]

Hi Piyush,

if you explicitly set a trigger, the default trigger of the window is replaced.

In your example, the time trigger is replaced by the count trigger, i.e., the window is only evaluated after the 100th element was received.

This blog post discusses windows and triggers [1].

Best, Fabian

[1] http://flink.apache.org/news/2015/12/04/Introducing-windows.html

2016-04-20 13:20 GMT+02:00 Piyush Shrivastava <[hidden email]>:

I wanted to know how Windows and Triggers work in Flink. I am creating a time window of 20 seconds and a count trigger of 100.

stream.keyBy(0)
             .timeWindow(Time.seconds(20))
             .trigger(CountTrigger.of(100))

In this case, when will my window get triggered? When 20 seconds has passed, 100 messages are passed?

Thanks and Regards,

[hidden email]

http://webograffiti.com

[Piyush Shrivastava]

Hi Fabian,

Thanks for the information. I also quickly want to ask that if I implement a custom trigger that fires in one hour for the first time and then every five minutes, what all functions do I need to use?

I am considering creating my own trigger referring the code here:

https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java

What changes do I need to make? Is it even possible to do this?

 

On Wednesday, 20 April 2016 4:59 PM, Fabian Hueske <[hidden email]> wrote:

Hi Piyush,

if you explicitly set a trigger, the default trigger of the window is replaced.

In your example, the time trigger is replaced by the count trigger, i.e., the window is only evaluated after the 100th element was received.

This blog post discusses windows and triggers [1].

Best, Fabian

[1] http://flink.apache.org/news/2015/12/04/Introducing-windows.html

2016-04-20 13:20 GMT+02:00 Piyush Shrivastava <[hidden email]>:

I wanted to know how Windows and Triggers work in Flink. I am creating a time window of 20 seconds and a count trigger of 100.

stream.keyBy(0)
             .timeWindow(Time.seconds(20))
             .trigger(CountTrigger.of(100))

In this case, when will my window get triggered? When 20 seconds has passed, 100 messages are passed?

Thanks and Regards,

[hidden email]

http://webograffiti.com

[Fabian Hueske-2]

Hi Piyush,

that's not trivial to implement. You can only do that with a so-called GlobalWindow, i.e., a window which receives all elements of a partition, and a custom trigger which has state to decide whether it has triggered the first window or not. It won't work with a CountTrigger.

Best, Fabian

2016-04-20 14:20 GMT+02:00 Piyush Shrivastava <[hidden email]>:

Hi Fabian,

Thanks for the information. I also quickly want to ask that if I implement a custom trigger that fires in one hour for the first time and then every five minutes, what all functions do I need to use?

I am considering creating my own trigger referring the code here:

https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java

What changes do I need to make? Is it even possible to do this?

 

Thanks and Regards,

[hidden email]

http://webograffiti.com

On Wednesday, 20 April 2016 4:59 PM, Fabian Hueske <[hidden email]> wrote:

Hi Piyush,

if you explicitly set a trigger, the default trigger of the window is replaced.

In your example, the time trigger is replaced by the count trigger, i.e., the window is only evaluated after the 100th element was received.

This blog post discusses windows and triggers [1].

Best, Fabian

[1] http://flink.apache.org/news/2015/12/04/Introducing-windows.html

2016-04-20 13:20 GMT+02:00 Piyush Shrivastava <[hidden email]>:

I wanted to know how Windows and Triggers work in Flink. I am creating a time window of 20 seconds and a count trigger of 100.

stream.keyBy(0)
             .timeWindow(Time.seconds(20))
             .trigger(CountTrigger.of(100))

In this case, when will my window get triggered? When 20 seconds has passed, 100 messages are passed?

Thanks and Regards,

[hidden email]

http://webograffiti.com

[Aljoscha Krettek]

Just to clarify, the state of a Trigger on GlobalWindows is still local to the key of the element that is in the window(s).

On Wed, 20 Apr 2016 at 18:11 Fabian Hueske <[hidden email]> wrote:

Hi Piyush,

that's not trivial to implement. You can only do that with a so-called GlobalWindow, i.e., a window which receives all elements of a partition, and a custom trigger which has state to decide whether it has triggered the first window or not. It won't work with a CountTrigger.

Best, Fabian

2016-04-20 14:20 GMT+02:00 Piyush Shrivastava <[hidden email]>:

Hi Fabian,

Thanks for the information. I also quickly want to ask that if I implement a custom trigger that fires in one hour for the first time and then every five minutes, what all functions do I need to use?

I am considering creating my own trigger referring the code here:

https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java

What changes do I need to make? Is it even possible to do this?

 

Thanks and Regards,

[hidden email]

http://webograffiti.com

On Wednesday, 20 April 2016 4:59 PM, Fabian Hueske <[hidden email]> wrote:

Hi Piyush,

if you explicitly set a trigger, the default trigger of the window is replaced.

In your example, the time trigger is replaced by the count trigger, i.e., the window is only evaluated after the 100th element was received.

This blog post discusses windows and triggers [1].

Best, Fabian

[1] http://flink.apache.org/news/2015/12/04/Introducing-windows.html

2016-04-20 13:20 GMT+02:00 Piyush Shrivastava <[hidden email]>:

I wanted to know how Windows and Triggers work in Flink. I am creating a time window of 20 seconds and a count trigger of 100.

stream.keyBy(0)
             .timeWindow(Time.seconds(20))
             .trigger(CountTrigger.of(100))

In this case, when will my window get triggered? When 20 seconds has passed, 100 messages are passed?

Thanks and Regards,

[hidden email]

http://webograffiti.com

[Piyush Shrivastava]

Hi all,

Thanks a lot for your valuable suggestions. I am trying to implement the logic of creating a custom trigger with a GlobalWindow which fires the window in one minute for the first time and every five seconds after that, if this logic works I will change it to one hour and five minutes respectively.

@Fabian, it might not be very trivial to implement but is required in my particular use case as I need the data from the first one hour to start operating on, after that I want to append data in every five minutes.

In my custom trigger, the code looks like this:

@Override
    public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
               
        if(flag == false){
            if(i<60){
                Thread.sleep(1000);
                System.out.println("flag: "+flag+" count: "+i);
                i++;
                return TriggerResult.CONTINUE;
            }
            flag = true;
            return TriggerResult.FIRE;
        }else{
            if(j<5){
                Thread.sleep(1000);
                System.out.println("flag: "+flag+" count: "+j);
                j++;
                return TriggerResult.CONTINUE;
            }
            j=0;
            return TriggerResult.FIRE;
        }
           
    }

 

Here, flag is a static boolean which is false for the first run and true for the next runs. i and j are static counters initialized at 0. However, when I run the stream, only one or two results (one or two keys) are coming in the output every time the Window is fired. Can you figure out why this is happening? How can I correct this logic so that all the keys are considered.

Sample of the output I am getting:

flag: false count: 0
flag: false count: 1
flag: false count: 2
flag: false count: 3
flag: false count: 4
flag: false count: 5
flag: false count: 6
flag: false count: 7
flag: false count: 8
flag: false count: 9
flag: false count: 10
.

.

.

flag: false count: 58
flag: false count: 58
3> (52,"mokshda",85.929)
flag: true count: 60
flag: true count: 61
flag: true count: 0
flag: true count: 1
flag: true count: 2
flag: true count: 3
flag: true count: 4
1> (2,"poorvi",23.15)
flag: true count: 0
flag: true count: 1
flag: true count: 2
flag: true count: 3
flag: true count: 3
3> (52,"mokshda",85.930)
1> (2,"poorvi",23.15)
flag: true count: 0
flag: true count: 1
flag: true count: 2

.

.

.

On Wednesday, 20 April 2016 10:26 PM, Aljoscha Krettek <[hidden email]> wrote:

Just to clarify, the state of a Trigger on GlobalWindows is still local to the key of the element that is in the window(s).

On Wed, 20 Apr 2016 at 18:11 Fabian Hueske <[hidden email]> wrote:

Hi Piyush,

that's not trivial to implement. You can only do that with a so-called GlobalWindow, i.e., a window which receives all elements of a partition, and a custom trigger which has state to decide whether it has triggered the first window or not. It won't work with a CountTrigger.

Best, Fabian

2016-04-20 14:20 GMT+02:00 Piyush Shrivastava <[hidden email]>:

Hi Fabian,

Thanks for the information. I also quickly want to ask that if I implement a custom trigger that fires in one hour for the first time and then every five minutes, what all functions do I need to use?

I am considering creating my own trigger referring the code here:

https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/triggers/CountTrigger.java

What changes do I need to make? Is it even possible to do this?

 

Thanks and Regards,

[hidden email]

http://webograffiti.com

On Wednesday, 20 April 2016 4:59 PM, Fabian Hueske <[hidden email]> wrote:

Hi Piyush,

if you explicitly set a trigger, the default trigger of the window is replaced.

In your example, the time trigger is replaced by the count trigger, i.e., the window is only evaluated after the 100th element was received.

This blog post discusses windows and triggers [1].

Best, Fabian

[1] http://flink.apache.org/news/2015/12/04/Introducing-windows.html

2016-04-20 13:20 GMT+02:00 Piyush Shrivastava <[hidden email]>:

I wanted to know how Windows and Triggers work in Flink. I am creating a time window of 20 seconds and a count trigger of 100.

stream.keyBy(0)
             .timeWindow(Time.seconds(20))
             .trigger(CountTrigger.of(100))

In this case, when will my window get triggered? When 20 seconds has passed, 100 messages are passed?

Thanks and Regards,

[hidden email]

http://webograffiti.com