Flink Window with multiple trigger condition

[anuj.aj07]

Hello All,

I am getting a lot of user events in a  stream. There are different types of events, now I want to build some aggregation metrics for the user by grouping events in buckets.

My condition for windowing is :

1. Start the window for the user when event_name: "search"  arrived for the user.

2. Trigger the window when 

      either 30 mins from the start of the window reached 

       OR

       event_type : "start" is appeared.

After that, I want to do calculate some aggregation on those window events. I know this can be done using process function but I am stuck to create the window with multiple conditions trigger.

Please help me how to create this type of window with multiple trigger condition either time or some event happen.

--

[anuj.aj07]

Session window defined on the gap of inactivity, I do not have that requirement. 

Start the window only on the "search event" that part I will take later. 

Let's say in the first phase I want to start the window on any event that appears for that user. 

For example :

Scenario -1

t1 ----- user1  ---- event1 ( window start)

t1 +5 mins ----- user1 ----- event2 

t1 + 10 mins --- user1 ---- event3

t1 + 15 mins ----- user1 ---- event4===start type event (terminate window as event type "Start" arrived and calculate aggregate on above collected events)

t1+16 mins ---user-1  ---- event 5 starts a new window

Scenario -2

t1 ----- user1  ---- event1 ( window start)

t1 +5 mins ----- user1 ----- event2 

t1 + 10 mins --- user1 ---- event3

t1 + 30 mins ----- user1 ---- event4 (terminates the window as 30 mins elapsed and calculate aggregate on above collected events)

t1+31 mins ---user-1  ---- event5  starts a new window

This I want to implement. I have tried to read triggers but did not getting understand how to trigger when either time pass or eventtype== "start" has arrived.  Which function of trigger class I have to implement and how to check these 2  conditions on each event arrive.  

Please help to implement this. If you can provide a basic start function that I need to implement. I am not clear how to start.

On Thu, May 21, 2020 at 4:59 PM Jiayi Liao <[hidden email]> wrote:

According to your description, it's more like a session window scenario rather than a tumbling window, you can find more details in [1]. But the difference between your window and Flink

's session window is, the session length is defined by the first element in your case. I'm afraid that Flink does't have implementations for such scenarios, you may need to create your own WindowAssigner.

For the trigger, yes, you can always implement a trigger to determine the lifecyle of a window.

[1]. https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/windows.html#session-windows

Best,

Jiayi Liao

On Thu, May 21, 2020 at 5:51 PM aj <[hidden email]> wrote:

Hi Liao, 

Thanks for the response. I am reading all this as I never implemented this. 

> Start the window for the user when event_name: "search"  arrived for the user.

I'm not very sure this is right way to define the window in your business if you use event time, because there may exist out-of-order events that happened after "search" event, but arrive before "search" event, which will be discarded because you haven't assigned a window. (If I understand correctly)

Yes you are right and I raised this concern to the business team and we still in discussion. 

But let say if I do not need the above condition if I want to start the window whenever the first event of the particular user event appears and then bucket those events with similar conditions (either 30 mins from the start of the window reached or event_type: "start" is appeared). So, in that case, can I use TumblingProcessingTimeWindows with 30 mins, and on that can I put a custom trigger that before 30 mins if event_type: "start" is arrived than the process the window.

Is this correct understanding like if let stay start event arrived at 20 mins from window start then that window will be close and processed and events that arriving after that will be assign to the new window or window will continue till 30 mins.

On Thu, May 21, 2020 at 2:55 PM Jiayi Liao <[hidden email]> wrote:

Hi,

> Start the window for the user when event_name: "search"  arrived for the user.

I'm not very sure this is right way to define the window in your business if you use event time, because there may exist out-of-order events that happened after "search" event, but arrive before "search" event, which will be discarded because you haven't assigned a window. (If I understand correctly)

Back to the problem, I think you can implement your own #WindowAssigner, which will create a window based on the event's name. Take a look at our existing implementations like #TumblingEventWindows.

> Trigger the window when either 30 mins from the start of the window reached or event_type : "start" is appeared

This can also be implemented with a customed #Trigger. The timing of being triggered can be set by registering timers with Flink's internal timer service. Take a look at #EventTimeTrigger, it's easy to implement it.

Best,

Jiayi Liao

On Thu, May 21, 2020 at 2:47 PM aj <[hidden email]> wrote:

Hello All,

I am getting a lot of user events in a  stream. There are different types of events, now I want to build some aggregation metrics for the user by grouping events in buckets.

My condition for windowing is :

1. Start the window for the user when event_name: "search"  arrived for the user.

2. Trigger the window when 

      either 30 mins from the start of the window reached 

       OR

       event_type : "start" is appeared.

After that, I want to do calculate some aggregation on those window events. I know this can be done using process function but I am stuck to create the window with multiple conditions trigger.

Please help me how to create this type of window with multiple trigger condition either time or some event happen.

--

Thanks & Regards,
Anuj Jain

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

[Tzu-Li (Gordon) Tai]

Hi,

To achieve what you have in mind, I think what you have to do is to use a
processing time window of 30 mins, and have a custom trigger that matches
the "start" event in the `onElement` method and return
TriggerResult.FIRE_AND_PURGE.

That way, the window fires either when the processing time has passed, or
the start event was recieved.

Cheers,
Gordon



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[anuj.aj07]

 

I was also thinking to have a processing time window but that will not work for me. I want to start the window when the user  "search" event arrives. So for each user window will start from the search event. 

 The Tumbling window has fixed start end time so that will not be suitable in my case. 

On Fri, May 22, 2020 at 10:23 AM Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi,

To achieve what you have in mind, I think what you have to do is to use a
processing time window of 30 mins, and have a custom trigger that matches
the "start" event in the `onElement` method and return
TriggerResult.FIRE_AND_PURGE.

That way, the window fires either when the processing time has passed, or
the start event was recieved.

Cheers,
Gordon



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

[anuj.aj07]

I am still not able to get much after reading the stuff. Please help with some basic code to start to build this window and trigger. 

Another option I am thinking is I just use a Richflatmap function and use the keyed state to build this logic. Is that the correct approach? 

On Fri, May 22, 2020 at 4:52 PM aj <[hidden email]> wrote:

 

I was also thinking to have a processing time window but that will not work for me. I want to start the window when the user  "search" event arrives. So for each user window will start from the search event. 

 The Tumbling window has fixed start end time so that will not be suitable in my case. 

On Fri, May 22, 2020 at 10:23 AM Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi,

To achieve what you have in mind, I think what you have to do is to use a
processing time window of 30 mins, and have a custom trigger that matches
the "start" event in the `onElement` method and return
TriggerResult.FIRE_AND_PURGE.

That way, the window fires either when the processing time has passed, or
the start event was recieved.

Cheers,
Gordon



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

[Yun Gao]

Hi,

   

       First sorry that I'm not expert on Window and please correct me if I'm wrong, but from my side, it seems the assigner might also be a problem in addition to the trigger: currently Flink window assigner should be all based on time (processing time or event time), and it might be hard to implement an event-driven window assigner that start to assign elements to a window after received some elements. 

   

      What comes to me is that a possible alternative method is to use the low-level KeyedProcessFunction directly:  you may register a timer 30 mins later when received the "search" event and write the time of search event into the state. Then for the following events, they will be saved to the state since the flag is set. After received the "start" event or the timer is triggered, you could load all the events from the states, do the aggregation and cancel the timer if it is triggered by "start" event. A simpler case is [1] and it does not consider stop the aggreation when received special event, but it seems that the logic could be added to the case.

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/process_function.html#example

Best,

 Yun

------------------Original Mail ------------------

Sender:aj <[hidden email]>

Send Date:Sun May 24 01:10:55 2020

Recipients:Tzu-Li (Gordon) Tai <[hidden email]>

CC:user <[hidden email]>

Subject:Re: Flink Window with multiple trigger condition

I am still not able to get much after reading the stuff. Please help with some basic code to start to build this window and trigger. 

Another option I am thinking is I just use a Richflatmap function and use the keyed state to build this logic. Is that the correct approach? 

On Fri, May 22, 2020 at 4:52 PM aj <[hidden email]> wrote:

 

I was also thinking to have a processing time window but that will not work for me. I want to start the window when the user  "search" event arrives. So for each user window will start from the search event. 

 The Tumbling window has fixed start end time so that will not be suitable in my case. 

On Fri, May 22, 2020 at 10:23 AM Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi,

To achieve what you have in mind, I think what you have to do is to use a
processing time window of 30 mins, and have a custom trigger that matches
the "start" event in the `onElement` method and return
TriggerResult.FIRE_AND_PURGE.

That way, the window fires either when the processing time has passed, or
the start event was recieved.

Cheers,
Gordon



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

[anuj.aj07]

Hi,

I have implemented the below solution and its working fine but the biggest problem with this is if no event coming for the user after 30 min then I am not able to trigger because I am checking

time diff from upcoming events. So when the next event comes than only it triggers but I want it to trigger just after 30 mins. 

So please help me to improve this and how to solve the above problem.

public class DemandSessionFlatMap extends RichFlatMapFunction<Tuple2<Long, GenericRecord>, DemandSessionSummaryTuple> {

    private static final Logger LOGGER = LoggerFactory.getLogger(DemandSessionFlatMap.class);

    private transient ValueState<Tuple3<String, Long, Long>> timeState; // maintain session_id starttime and endtime 
    private transient MapState<String, DemandSessionSummaryTuple> sessionSummary; // map for hex9 and summarytuple

    @Override
    public void open(Configuration config) {

        ValueStateDescriptor<Tuple3<String, Long, Long>> timeDescriptor =
                new ValueStateDescriptor<>(
                        "time_state", // the state name
                        TypeInformation.of(new TypeHint<Tuple3<String, Long, Long>>() {
                        }), // type information
                        Tuple3.of(null, 0L, 0L)); // default value of the state, if nothing was set
        timeState = getRuntimeContext().getState(timeDescriptor);

        MapStateDescriptor<String, DemandSessionSummaryTuple> descriptor =
                new MapStateDescriptor<String, DemandSessionSummaryTuple>("demand_session",
                        TypeInformation.of(new TypeHint<String>() {
                        }), TypeInformation.of(new TypeHint<DemandSessionSummaryTuple>() {
                }));
        sessionSummary = getRuntimeContext().getMapState(descriptor);

    }

    @Override
    public void flatMap(Tuple2<Long, GenericRecord> recordTuple2, Collector<DemandSessionSummaryTuple> collector) throws Exception {
        GenericRecord record = recordTuple2.f1;
        String event_name = record.get("event_name").toString();
        long event_ts = (Long) record.get("event_ts");
        Tuple3<String, Long, Long> currentTimeState = timeState.value();

        if (event_name.equals("search_list_keyless") && currentTimeState.f1 == 0) {
            currentTimeState.f1 = event_ts;
            String demandSessionId = UUID.randomUUID().toString();
            currentTimeState.f0 = demandSessionId;
        }

        long timeDiff = event_ts - currentTimeState.f1;

        if (event_name.equals("keyless_start_trip") || timeDiff >= 1800000) {
            Tuple3<String, Long, Long> finalCurrentTimeState = currentTimeState;
            sessionSummary.entries().forEach( tuple ->{
                String key = tuple.getKey();
                DemandSessionSummaryTuple sessionSummaryTuple = tuple.getValue();
                try {
                    sessionSummaryTuple.setEndTime(finalCurrentTimeState.f2);
                    collector.collect(sessionSummaryTuple);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            });
            timeState.clear();
            sessionSummary.clear();
            currentTimeState = timeState.value();
        }

        if (event_name.equals("search_list_keyless") && currentTimeState.f1 == 0) {
            currentTimeState.f1 = event_ts;
            String demandSessionId = UUID.randomUUID().toString();
            currentTimeState.f0 = demandSessionId;
        }
        currentTimeState.f2 = event_ts;

        if (currentTimeState.f1 > 0) {
            String search_hex9 = record.get("search_hex9") != null ? record.get("search_hex9").toString() : null;
            DemandSessionSummaryTuple currentTuple = sessionSummary.get(search_hex9) != null ? sessionSummary.get(search_hex9) : new DemandSessionSummaryTuple();

            if (sessionSummary.get(search_hex9) == null) {
                currentTuple.setSearchHex9(search_hex9);
                currentTuple.setUserId(recordTuple2.f0);
                currentTuple.setStartTime(currentTimeState.f1);
                currentTuple.setDemandSessionId(currentTimeState.f0);
            }

            if (event_name.equals("search_list_keyless")) {
                currentTuple.setTotalSearch(currentTuple.getTotalSearch() + 1);
                SearchSummaryCalculation(record, currentTuple);
            }
            sessionSummary.put(search_hex9, currentTuple);
        }
        timeState.update(currentTimeState);
    }

On Sun, May 24, 2020 at 10:57 PM Yun Gao <[hidden email]> wrote:

Hi,

   

       First sorry that I'm not expert on Window and please correct me if I'm wrong, but from my side, it seems the assigner might also be a problem in addition to the trigger: currently Flink window assigner should be all based on time (processing time or event time), and it might be hard to implement an event-driven window assigner that start to assign elements to a window after received some elements. 

   

      What comes to me is that a possible alternative method is to use the low-level KeyedProcessFunction directly:  you may register a timer 30 mins later when received the "search" event and write the time of search event into the state. Then for the following events, they will be saved to the state since the flag is set. After received the "start" event or the timer is triggered, you could load all the events from the states, do the aggregation and cancel the timer if it is triggered by "start" event. A simpler case is [1] and it does not consider stop the aggreation when received special event, but it seems that the logic could be added to the case.

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/process_function.html#example

Best,

 Yun

------------------Original Mail ------------------

Sender:aj <[hidden email]>

Send Date:Sun May 24 01:10:55 2020

Recipients:Tzu-Li (Gordon) Tai <[hidden email]>

CC:user <[hidden email]>

Subject:Re: Flink Window with multiple trigger condition

I am still not able to get much after reading the stuff. Please help with some basic code to start to build this window and trigger. 

Another option I am thinking is I just use a Richflatmap function and use the keyed state to build this logic. Is that the correct approach? 

On Fri, May 22, 2020 at 4:52 PM aj <[hidden email]> wrote:

 

I was also thinking to have a processing time window but that will not work for me. I want to start the window when the user  "search" event arrives. So for each user window will start from the search event. 

 The Tumbling window has fixed start end time so that will not be suitable in my case. 

On Fri, May 22, 2020 at 10:23 AM Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi,

To achieve what you have in mind, I think what you have to do is to use a
processing time window of 30 mins, and have a custom trigger that matches
the "start" event in the `onElement` method and return
TriggerResult.FIRE_AND_PURGE.

That way, the window fires either when the processing time has passed, or
the start event was recieved.

Cheers,
Gordon



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

[Yun Gao]

Hi,

    

     I think you could use timer to achieve that. In processFunction you could register a timer at specific time (event time or processing time) and get callbacked at that point. It could be registered like 

    

ctx.timerService().registerEventTimeTimer(current.lastModified + 60000);

    

    More details on timer could be found in [1] and an example is in [2]. In this example, a timer is registered in the last line of the processElement method, and the callback is implemented by override the onTimer method.

   [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/stream/operators/process_function.html#timers

   [2] https://ci.apache.org/projects/flink/flink-docs-master/dev/stream/operators/process_function.html#example

------------------Original Mail ------------------

Sender:aj <[hidden email]>

Send Date:Fri May 29 02:07:33 2020

Recipients:Yun Gao <[hidden email]>

CC:user <[hidden email]>

Subject:Re: Re: Flink Window with multiple trigger condition

Hi,

I have implemented the below solution and its working fine but the biggest problem with this is if no event coming for the user after 30 min then I am not able to trigger because I am checking

time diff from upcoming events. So when the next event comes than only it triggers but I want it to trigger just after 30 mins. 

So please help me to improve this and how to solve the above problem.

public class DemandSessionFlatMap extends RichFlatMapFunction<Tuple2<Long, GenericRecord>, DemandSessionSummaryTuple> {

    private static final Logger LOGGER = LoggerFactory.getLogger(DemandSessionFlatMap.class);

    private transient ValueState<Tuple3<String, Long, Long>> timeState; // maintain session_id starttime and endtime 
    private transient MapState<String, DemandSessionSummaryTuple> sessionSummary; // map for hex9 and summarytuple

    @Override
    public void open(Configuration config) {

        ValueStateDescriptor<Tuple3<String, Long, Long>> timeDescriptor =
                new ValueStateDescriptor<>(
                        "time_state", // the state name
                        TypeInformation.of(new TypeHint<Tuple3<String, Long, Long>>() {
                        }), // type information
                        Tuple3.of(null, 0L, 0L)); // default value of the state, if nothing was set
        timeState = getRuntimeContext().getState(timeDescriptor);

        MapStateDescriptor<String, DemandSessionSummaryTuple> descriptor =
                new MapStateDescriptor<String, DemandSessionSummaryTuple>("demand_session",
                        TypeInformation.of(new TypeHint<String>() {
                        }), TypeInformation.of(new TypeHint<DemandSessionSummaryTuple>() {
                }));
        sessionSummary = getRuntimeContext().getMapState(descriptor);

    }

    @Override
    public void flatMap(Tuple2<Long, GenericRecord> recordTuple2, Collector<DemandSessionSummaryTuple> collector) throws Exception {
        GenericRecord record = recordTuple2.f1;
        String event_name = record.get("event_name").toString();
        long event_ts = (Long) record.get("event_ts");
        Tuple3<String, Long, Long> currentTimeState = timeState.value();

        if (event_name.equals("search_list_keyless") && currentTimeState.f1 == 0) {
            currentTimeState.f1 = event_ts;
            String demandSessionId = UUID.randomUUID().toString();
            currentTimeState.f0 = demandSessionId;
        }

        long timeDiff = event_ts - currentTimeState.f1;

        if (event_name.equals("keyless_start_trip") || timeDiff >= 1800000) {
            Tuple3<String, Long, Long> finalCurrentTimeState = currentTimeState;
            sessionSummary.entries().forEach( tuple ->{
                String key = tuple.getKey();
                DemandSessionSummaryTuple sessionSummaryTuple = tuple.getValue();
                try {
                    sessionSummaryTuple.setEndTime(finalCurrentTimeState.f2);
                    collector.collect(sessionSummaryTuple);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            });
            timeState.clear();
            sessionSummary.clear();
            currentTimeState = timeState.value();
        }

        if (event_name.equals("search_list_keyless") && currentTimeState.f1 == 0) {
            currentTimeState.f1 = event_ts;
            String demandSessionId = UUID.randomUUID().toString();
            currentTimeState.f0 = demandSessionId;
        }
        currentTimeState.f2 = event_ts;

        if (currentTimeState.f1 > 0) {
            String search_hex9 = record.get("search_hex9") != null ? record.get("search_hex9").toString() : null;
            DemandSessionSummaryTuple currentTuple = sessionSummary.get(search_hex9) != null ? sessionSummary.get(search_hex9) : new DemandSessionSummaryTuple();

            if (sessionSummary.get(search_hex9) == null) {
                currentTuple.setSearchHex9(search_hex9);
                currentTuple.setUserId(recordTuple2.f0);
                currentTuple.setStartTime(currentTimeState.f1);
                currentTuple.setDemandSessionId(currentTimeState.f0);
            }

            if (event_name.equals("search_list_keyless")) {
                currentTuple.setTotalSearch(currentTuple.getTotalSearch() + 1);
                SearchSummaryCalculation(record, currentTuple);
            }
            sessionSummary.put(search_hex9, currentTuple);
        }
        timeState.update(currentTimeState);
    }

On Sun, May 24, 2020 at 10:57 PM Yun Gao <[hidden email]> wrote:

Hi,

   

       First sorry that I'm not expert on Window and please correct me if I'm wrong, but from my side, it seems the assigner might also be a problem in addition to the trigger: currently Flink window assigner should be all based on time (processing time or event time), and it might be hard to implement an event-driven window assigner that start to assign elements to a window after received some elements. 

   

      What comes to me is that a possible alternative method is to use the low-level KeyedProcessFunction directly:  you may register a timer 30 mins later when received the "search" event and write the time of search event into the state. Then for the following events, they will be saved to the state since the flag is set. After received the "start" event or the timer is triggered, you could load all the events from the states, do the aggregation and cancel the timer if it is triggered by "start" event. A simpler case is [1] and it does not consider stop the aggreation when received special event, but it seems that the logic could be added to the case.

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/process_function.html#example

Best,

 Yun

------------------Original Mail ------------------

Sender:aj <[hidden email]>

Send Date:Sun May 24 01:10:55 2020

Recipients:Tzu-Li (Gordon) Tai <[hidden email]>

CC:user <[hidden email]>

Subject:Re: Flink Window with multiple trigger condition

I am still not able to get much after reading the stuff. Please help with some basic code to start to build this window and trigger. 

Another option I am thinking is I just use a Richflatmap function and use the keyed state to build this logic. Is that the correct approach? 

On Fri, May 22, 2020 at 4:52 PM aj <[hidden email]> wrote:

 

I was also thinking to have a processing time window but that will not work for me. I want to start the window when the user  "search" event arrives. So for each user window will start from the search event. 

 The Tumbling window has fixed start end time so that will not be suitable in my case. 

On Fri, May 22, 2020 at 10:23 AM Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi,

To achieve what you have in mind, I think what you have to do is to use a
processing time window of 30 mins, and have a custom trigger that matches
the "start" event in the `onElement` method and return
TriggerResult.FIRE_AND_PURGE.

That way, the window fires either when the processing time has passed, or
the start event was recieved.

Cheers,
Gordon



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

[anuj.aj07]

Thanks Yun.

I have converted the code to use a keyed-processed function rather than a flatMap and using register timer it worked. 

On Fri, May 29, 2020 at 11:13 AM Yun Gao <[hidden email]> wrote:

Hi,

    

     I think you could use timer to achieve that. In processFunction you could register a timer at specific time (event time or processing time) and get callbacked at that point. It could be registered like 

    

ctx.timerService().registerEventTimeTimer(current.lastModified + 60000);

    

    More details on timer could be found in [1] and an example is in [2]. In this example, a timer is registered in the last line of the processElement method, and the callback is implemented by override the onTimer method.

   [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/stream/operators/process_function.html#timers

   [2] https://ci.apache.org/projects/flink/flink-docs-master/dev/stream/operators/process_function.html#example

------------------Original Mail ------------------

Sender:aj <[hidden email]>

Send Date:Fri May 29 02:07:33 2020

Recipients:Yun Gao <[hidden email]>

CC:user <[hidden email]>

Subject:Re: Re: Flink Window with multiple trigger condition

Hi,

I have implemented the below solution and its working fine but the biggest problem with this is if no event coming for the user after 30 min then I am not able to trigger because I am checking

time diff from upcoming events. So when the next event comes than only it triggers but I want it to trigger just after 30 mins. 

So please help me to improve this and how to solve the above problem.

public class DemandSessionFlatMap extends RichFlatMapFunction<Tuple2<Long, GenericRecord>, DemandSessionSummaryTuple> {

    private static final Logger LOGGER = LoggerFactory.getLogger(DemandSessionFlatMap.class);

    private transient ValueState<Tuple3<String, Long, Long>> timeState; // maintain session_id starttime and endtime 
    private transient MapState<String, DemandSessionSummaryTuple> sessionSummary; // map for hex9 and summarytuple

    @Override
    public void open(Configuration config) {

        ValueStateDescriptor<Tuple3<String, Long, Long>> timeDescriptor =
                new ValueStateDescriptor<>(
                        "time_state", // the state name
                        TypeInformation.of(new TypeHint<Tuple3<String, Long, Long>>() {
                        }), // type information
                        Tuple3.of(null, 0L, 0L)); // default value of the state, if nothing was set
        timeState = getRuntimeContext().getState(timeDescriptor);

        MapStateDescriptor<String, DemandSessionSummaryTuple> descriptor =
                new MapStateDescriptor<String, DemandSessionSummaryTuple>("demand_session",
                        TypeInformation.of(new TypeHint<String>() {
                        }), TypeInformation.of(new TypeHint<DemandSessionSummaryTuple>() {
                }));
        sessionSummary = getRuntimeContext().getMapState(descriptor);

    }

    @Override
    public void flatMap(Tuple2<Long, GenericRecord> recordTuple2, Collector<DemandSessionSummaryTuple> collector) throws Exception {
        GenericRecord record = recordTuple2.f1;
        String event_name = record.get("event_name").toString();
        long event_ts = (Long) record.get("event_ts");
        Tuple3<String, Long, Long> currentTimeState = timeState.value();

        if (event_name.equals("search_list_keyless") && currentTimeState.f1 == 0) {
            currentTimeState.f1 = event_ts;
            String demandSessionId = UUID.randomUUID().toString();
            currentTimeState.f0 = demandSessionId;
        }

        long timeDiff = event_ts - currentTimeState.f1;

        if (event_name.equals("keyless_start_trip") || timeDiff >= 1800000) {
            Tuple3<String, Long, Long> finalCurrentTimeState = currentTimeState;
            sessionSummary.entries().forEach( tuple ->{
                String key = tuple.getKey();
                DemandSessionSummaryTuple sessionSummaryTuple = tuple.getValue();
                try {
                    sessionSummaryTuple.setEndTime(finalCurrentTimeState.f2);
                    collector.collect(sessionSummaryTuple);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            });
            timeState.clear();
            sessionSummary.clear();
            currentTimeState = timeState.value();
        }

        if (event_name.equals("search_list_keyless") && currentTimeState.f1 == 0) {
            currentTimeState.f1 = event_ts;
            String demandSessionId = UUID.randomUUID().toString();
            currentTimeState.f0 = demandSessionId;
        }
        currentTimeState.f2 = event_ts;

        if (currentTimeState.f1 > 0) {
            String search_hex9 = record.get("search_hex9") != null ? record.get("search_hex9").toString() : null;
            DemandSessionSummaryTuple currentTuple = sessionSummary.get(search_hex9) != null ? sessionSummary.get(search_hex9) : new DemandSessionSummaryTuple();

            if (sessionSummary.get(search_hex9) == null) {
                currentTuple.setSearchHex9(search_hex9);
                currentTuple.setUserId(recordTuple2.f0);
                currentTuple.setStartTime(currentTimeState.f1);
                currentTuple.setDemandSessionId(currentTimeState.f0);
            }

            if (event_name.equals("search_list_keyless")) {
                currentTuple.setTotalSearch(currentTuple.getTotalSearch() + 1);
                SearchSummaryCalculation(record, currentTuple);
            }
            sessionSummary.put(search_hex9, currentTuple);
        }
        timeState.update(currentTimeState);
    }

On Sun, May 24, 2020 at 10:57 PM Yun Gao <[hidden email]> wrote:

Hi,

   

       First sorry that I'm not expert on Window and please correct me if I'm wrong, but from my side, it seems the assigner might also be a problem in addition to the trigger: currently Flink window assigner should be all based on time (processing time or event time), and it might be hard to implement an event-driven window assigner that start to assign elements to a window after received some elements. 

   

      What comes to me is that a possible alternative method is to use the low-level KeyedProcessFunction directly:  you may register a timer 30 mins later when received the "search" event and write the time of search event into the state. Then for the following events, they will be saved to the state since the flag is set. After received the "start" event or the timer is triggered, you could load all the events from the states, do the aggregation and cancel the timer if it is triggered by "start" event. A simpler case is [1] and it does not consider stop the aggreation when received special event, but it seems that the logic could be added to the case.

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/process_function.html#example

Best,

 Yun

------------------Original Mail ------------------

Sender:aj <[hidden email]>

Send Date:Sun May 24 01:10:55 2020

Recipients:Tzu-Li (Gordon) Tai <[hidden email]>

CC:user <[hidden email]>

Subject:Re: Flink Window with multiple trigger condition

I am still not able to get much after reading the stuff. Please help with some basic code to start to build this window and trigger. 

Another option I am thinking is I just use a Richflatmap function and use the keyed state to build this logic. Is that the correct approach? 

On Fri, May 22, 2020 at 4:52 PM aj <[hidden email]> wrote:

 

I was also thinking to have a processing time window but that will not work for me. I want to start the window when the user  "search" event arrives. So for each user window will start from the search event. 

 The Tumbling window has fixed start end time so that will not be suitable in my case. 

On Fri, May 22, 2020 at 10:23 AM Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi,

To achieve what you have in mind, I think what you have to do is to use a
processing time window of 30 mins, and have a custom trigger that matches
the "start" event in the `onElement` method and return
TriggerResult.FIRE_AND_PURGE.

That way, the window fires either when the processing time has passed, or
the start event was recieved.

Cheers,
Gordon



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--

Thanks & Regards,
Anuj Jain
Mob. : [phone]
Skype : anuj.jain07

--