Flip CEP skip policy aftermatchskipstrategy Skippastlastevent() matched no longer matches the Bikeng Guide

Indicates the skip policy after the matching process
Today is about flink cep How to achieve Scrolling between multiple windows
That is to avoid the following situations , Of course, whether you need to avoid it depends on your job needs or what you need to learn
flink cep pattern Code

// Define a matching pattern 
val pattern: Pattern[weather, weather] = Pattern
      .begin[weather]("one")
      // Once or more 
      .timesOrMore(12)
      // Time is 12 The range of hours   Equivalent to window 
      .within(Time.hours(12))

out -> {
    [1,May 10 15:00:00 ],[1,May 10 16:00:00 ],[1,May 10 17:00:00 ],[1,May 10 18:00:00 ]}
out -> {
    [1,May 10 16:00:00 ],[1,May 10 17:00:00 ],[1,May 10 18:00:00 ],[1,May 10 19:00:00 ]}
out -> {
    [1,May 10 17:00:00 ],[1,May 10 18:00:00 ],[1,May 10 19:00:00 ],[1,May 10 20:00:00 ]}

However, what you want is such a matching rule , After a match, this data is no longer used as the data source for other matches

out -> {
    [1,May 10 15:00:00 ],[1,May 10 16:00:00 ],[1,May 10 17:00:00 ],[1,May 10 18:00:00 ]}
out -> {
    [1,May 10 19:00:00 ],[1,May 10 20:00:00 ],[1,May 10 21:00:00 ],[1,May 10 22:00:00 ]}
out -> {
    [1,May 10 23:00:00 ],[1,May 11 00:00:00 ],[1,May 11 01:00:00 ],[1,May 11 02:00:00 ]}

Modify the matching pattern code

    /** * AfterMatchSkipStrategy  Indicates the skip policy after the matching process  */
     // Define a matching pattern 
    val pattern: Pattern[weather, weather] = Pattern
    //AfterMatchSkipStrategy  Indicates the skip policy after the matching process 
    //skipPastLastEvent  Discard every partial match that begins before the end of the issued match 
      .begin[weather]("one" , AfterMatchSkipStrategy.skipPastLastEvent())
            // Once or more 
      .timesOrMore(1)
      // Time is 12 The range of hours   Equivalent to window 
      .within(Time.hours(5))

Get hypothetical data

Then we can get the data we want Take a look at my actual data

==== Map(one -> Buffer(weather(Tue May 10 15:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0), weather(Tue May 10 16:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0), weather(Tue May 10 17:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0), weather(Tue May 10 18:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0), weather(Tue May 10 19:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0)))
 ---- Some(Buffer(weather(Tue May 10 15:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0), weather(Tue May 10 16:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0), weather(Tue May 10 17:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0), weather(Tue May 10 18:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0), weather(Tue May 10 19:00:00 CST 2022, Taiwan Province ,2428, Tainan ,9999.0,9999.0,9999.0)))
 ==== Map(one -> Buffer(weather(Tue May 10 15:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0), weather(Tue May 10 16:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0), weather(Tue May 10 17:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0), weather(Tue May 10 18:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0), weather(Tue May 10 19:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0)))
 ---- Some(Buffer(weather(Tue May 10 15:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0), weather(Tue May 10 16:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0), weather(Tue May 10 17:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0), weather(Tue May 10 18:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0), weather(Tue May 10 19:00:00 CST 2022, Taiwan Province ,2429, Taichung ,9999.0,9999.0,9999.0)))
 ==== Map(one -> Buffer(weather(Tue May 10 15:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0), weather(Tue May 10 16:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0), weather(Tue May 10 17:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0), weather(Tue May 10 18:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0), weather(Tue May 10 19:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0)))
 ---- Some(Buffer(weather(Tue May 10 15:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0), weather(Tue May 10 16:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0), weather(Tue May 10 17:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0), weather(Tue May 10 18:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0), weather(Tue May 10 19:00:00 CST 2022, Taiwan Province ,2426, Taipei ,9999.0,9999.0,9999.0)))
 ==== Map(one -> Buffer(weather(Tue May 10 15:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0), weather(Tue May 10 16:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0), weather(Tue May 10 17:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0), weather(Tue May 10 18:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0), weather(Tue May 10 19:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0)))
 ---- Some(Buffer(weather(Tue May 10 15:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0), weather(Tue May 10 16:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0), weather(Tue May 10 17:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0), weather(Tue May 10 18:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0), weather(Tue May 10 19:00:00 CST 2022, Taiwan Province ,2427, Taitung ,9999.0,9999.0,9999.0)))

There's danger ahead

The application of this method is actually problematic , Let's analyze
Our rule is that the matched data cannot be matched again , However, when we set .timesOrMore(1) We will find that there is only one piece of data at a time , Why is that ？
I analyze .timesOrMore(1) The pattern matching will make all the data that can be matched reach 1 Collect at times , If we add .begin[weather]("one",AfterMatchSkipStrategy.skipPastLastEvent()), We just learned this , It is to make all the matched data no longer match , This rule plus .timesOrMore(1) There will be one by default 1 Data matching , Therefore, all data matches are the same .！？

Key Bikeng

If we need to use .begin[weather]("one",AfterMatchSkipStrategy.skipPastLastEvent()) and .timesOrMore(1) Then we must consider at least a few pieces of data matching each time , In fact, the number of pieces has been fixed here , It is no longer or more matching rules , Here you can consider changing the number of matches dynamically , Database persistence can be done , Then real-time dynamic update . Dynamically change the number of fixed pieces to match , Here, due to lack of talent and learning , No better way has yet been found , If there is a solution in the next few days, I will update this blog .

The revolution has not yet succeeded - Comrades still have to work hard

Skip strategy

Next, we discuss the impact of different skip strategies on matching results ：

• Don't skip （NO_SKIP）
  Code calls AfterMatchSkipStrategy.noSkip(). This is the default strategy , All possible matches will be output . So the complete... Will be output here 6 A match .

• Skip to next （SKIP_TO_NEXT）
  Code calls AfterMatchSkipStrategy.skipToNext(). Find one a1 After the start of the maximum match , skip a1 All other matches at the beginning , Straight from the next a2 Start matching . Of course a2 The same is true. Skip other matches . The resulting （a1 a2 a3 b）,（a2 a3 b）,（a3 b）. You can see , This skip strategy is similar to using .greedy() The effect is the same .

• Skip all sub matches （SKIP_PAST_LAST_EVENT）
  Code calls AfterMatchSkipStrategy.skipPastLastEvent(). find a1 Start matching （a1 a2 a3 b） after , Skip all a1 until a3 The match at the beginning , It's equivalent to skipping all these sub matches . The resulting （a1 a2 a3 b）, This is the most streamlined skip strategy .

• Jump to the first （SKIP_TO_FIRST[a]）
  Code calls AfterMatchSkipStrategy.skipToFirst(“a”), Here you pass in a parameter , The first matching event that indicates which pattern to jump to . find a1 Start matching （a1 a2 a3 b） after , Jump to the first one a（ That is to say a1） Match for start , It's equivalent to leaving only a1 Start matching . The resulting （a1 a2 a3 b）,（a1 a2 b）,（a1 b）.

• Skip to the last （SKIP_TO_LAST[a]）
  Code calls AfterMatchSkipStrategy.skipToLast(“a”), Also pass in a parameter , Indicates the last matching event of which pattern to jump to . find a1 Start matching （a1 a2 a3 b） after , Skip all a1、a2 Start matching , Jump to the last a（ That is to say a3） Match for start . The resulting （a1 a2 a3 b）,（a3 b）.