Flink CEP 实践01,概念细节篇-未完

红太狼 2022-11-12 13:46 315阅读 0赞

# 1，Flink CEP 中 SimpleCondition/IterativeCondition的区别     #

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

应该是针对 个体模式 与 组合模式的用法区别，我在这里使用的感觉就效果是一样的。

# 2，flatselect 和 select的区别 #

我在这里使用的感觉就效果是一样的，有个可以输出多个结果，输出list，一个是只能输出单个实体。

# 3，consecutive 和 allowCombinations关键字 #

consecutive()显式指定它， 如果你想要非确定性的松弛连续性，你可以使用allowCombinations()方法

# 4，循环模式 times(x)不是严格紧邻 #

**对于循环模式（例如oneOrMore()和times()），默认是宽松的连续性。 如果你想要严格的连续性，你必须使用consecutive()显式指定它， 如果你想要非确定性的松弛连续性，你可以使用allowCombinations()方法。**

1.正常的单例模式

Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("start").where(new SimpleCondition<LoginEvent>() \{  
            @Override  
            public boolean filter(LoginEvent value) throws Exception \{  
                return "fail".equals(value.getEventType());  
            \}  
        \}).next("next").where(new SimpleCondition<LoginEvent>() \{  
            @Override  
            public boolean filter(LoginEvent value) throws Exception \{  
                return "fail".equals(value.getEventType());  
            \}    //正常的单例模式  
        \}).within(Time.seconds(2));  
  
2.如果换成循环模式：

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

运行结果与我想的不一样，不应该有下图红框，

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

添加关键字运行：

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5][]  
>

5，Flink state能否在 flatSelect 算子里面使用？

看起来不支持：

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6][]
> 
> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7][]

# 6，AfterMatchSkipStrategy关键字的使用 #

> **在给定的 Pattern 中，当同一事件符合多种模式条件组合之后，需要指定 AfterMatchSkipStrategy 策略以处理已经匹配的事件。在 AfterMatchSkipStrategy 配置中有四件事件处理策略，分别为 NO\_SKIP / SKIP\_PAST\_LAST\_EVENT / SKIP\_TO\_FIRST / SKIP\_TO\_LAST 。 每种策略的定义和使用方式如下：其中SKIP\_TO\_FIRST 和 SKIP\_TO\_LAST 在定义过程中需要指定有效的PatternName.**
> 
> ** NO\_SKIP: 该策略表示将所有可能匹配的事件进行输出，不忽略任何一条。**  
> AfterMatchSkipStrategy.noSkip()  
>   
> ** SKIP\_PAST\_LAST\_EVENT: 该策略表示忽略从模式条件开始触发到当前触发 Pattern 中的所有部分匹配事件。**  
> AfterMatchSkipStrategy.skipPastLastEvent()  
>   
> ** SKIP\_TO\_FIRST: 该策略表示忽略第一个匹配指定 PatternName 的 Pattern 其之前的部分匹配事件。**  
> AfterMatchSkipStrategy.skipToFirst(patternName)  
>   
> ** SKIP\_TO\_LAST 该策略表示忽略最后一个匹配指定 PatternName 的 Pattern 之前的部分匹配之间**  
> AfterMatchSkipStrategy.skipToLast(patternName)  
>   
> ** SKIP\_TO\_NEXT: 该策略表示忽略指定 PatternName 的 Pattern 之后的部分匹配事件**  
> AfterMatchSkipStrategy.skipToNext(patternName)  
>   
> 选择完 AfterMatchSkipStrategy 之后，可以再创建 Pattern 时 ， 通过 begin 方法中指定 skipStrategy , 然后就可以将 AfterMatchSkipStrategy 应用到当前的 Pattern 中。
> 
> val skipStrategy = \{ \}  
> Pattern.begin("pattern\_name" , skipStrategy)  
>

**实践操作应用：**

假定我们的数据是这样子的：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8][]

假定我们的代码是这样子的：

Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("begin")
    				.where(new RichIterativeCondition<JSONObject>() {
    					@Override
    					public boolean filter(JSONObject dataJson, Context<JSONObject> context) throws Exception {
    						Integer operation_type = dataJson.getInteger("operation_type");
    						if (operation_type==1){
    							return true;
    						}
    						return false;
    					}
    				}).next("middle")
    				.where(new RichIterativeCondition<JSONObject>() {
    					@Override
    					public boolean filter(JSONObject dataJson, Context<JSONObject> context) throws Exception {
    						//todo 这里中间如果不能是 operation_type=24
    						Integer operation_type = dataJson.getInteger("operation_type");
    						Integer event_id = dataJson.getInteger("event_id");
    
    						if (operation_type==1){
    							return true;
    						}
    						return false;
    					}
    				}).timesOrMore(2).consecutive()

我们正常得到的结果是：

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9][]

但是我不想得到这个结果，我想满足条件的情况下，一直以满足条件的第一条数据作为结果头：

添加代码：

**该策略表示忽略第一个匹配指定 PatternName 的 Pattern 其之前的部分匹配事件。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10][]

运行结果为：

> ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11][]

# 7，使用 within(Time.hours(1))  造成每次执行结果不一样   问题详情：使用eventTIme策略添加了warterMark之后，cep每次匹配结果输出条数不一样 。 #

“我在使用cep的时候 使用了within （window）指定时间策略为eventTime，然后运行输出结果不一样，差别还有点大，测试之后 发现使用 .assignTimestampsAndWatermarks(new AscendingTimestampExtractor 结果误差相对少一点。其他的watermark使用之后每次结果偏差都有点大。 我改为processTime策略之后发现结果输出都是一样的，那就排除了是window的问题。"

解决方案：1，checkpoint解决？  
                 2，通过监控kafka然后添加心跳数据？

# 8，使用waterMaker的遇到的问题，在flink1.12版本之前。 #

问题还是这一篇文章说的清楚：[https://article.itxueyuan.com/PgEMjO][https_article.itxueyuan.com_PgEMjO]

我遇到的问题是基本的cep代码，最后一条kafka数据触发不了计算，也是遇到的同样的问题，准备试试flink 1.12版本。

结论： 没卵用，不是解决这个问题的

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12][]

# 9，关于cep的一些概念问题 #

![preview][]

## **CEP-三种状态迁移边** ##

*  Take: 表示事件匹配成功，将当前状态更新到新状态，并前进到“下一个”状态；
 *  Procceed: 当事件来到的时候，当前状态不发生变化，在状态转换图中事件直接“前进”到下一个目标状态；
 *  IGNORE: 当事件来到的时候，如果匹配不成功，忽略当前事件，当前状态不发生任何变化。

为了更好的理解上述概念，我们利用下面代码，构建一个nfa：

//构建链接
    patterns Pattern pattern = Pattern.begin("start").where(new SimpleCondition() {
     private static final long serialVersionUID = 5726188262756267490L; 
    @Override public boolean filter(Event value) throws Exception { 
    return value.getName().equals("c");
                }
            }).followedBy("middle").where(new SimpleCondition() {
     private static final long serialVersionUID = 5726188262756267490L; 
    @Override public boolean filter(Event value) throws Exception { 
    return value.getName().equals("a");
                }
            }).optional(); 
    //创建nfa NFA nfa = NFACompiler.compile(pattern, Event.createTypeSerializer(), false);

例如构建的nfa的状态示意图如下所示：

![8ff1c609ccf6722c7af417dfc45b6510.png][]

此时如果我们加入数据源如下, 理论上的输出结果应该是\[event1\] 和 \[event1，event4\]。

Event event0 = new Event(40, "x", 1.0); 
    Event event1 = new Event(40, "c", 1.0); 
    Event event2 = new Event(42, "b", 2.0); 
    Event event3 = new Event(43, "b", 2.0); 
    Event event4 = new Event(44, "a", 2.0); 
    Event event5 = new Event(45, "b", 5.0);

下面我们来分析下，当第一条消息event0来的时候，由于start状态只有Take状态迁移边，这时event0匹配失败，消息被丢失，start状态不发生任何变化；当第二条消息event1来的时候，匹配成功，这时用event1更新start当前状态，并且进入下一个状态，既mid状态。而这是我们发现mid状态存在Proceed状态迁移边，以为着事件来临时，可以直接进入下一个状态，这里就是endstat状态，说明匹配结束，存在第一个匹配结果\[event1\]；当第三条消息event2来临时，由于之前我们已经进入了mid状态，所以nfa会让我们先匹配mid的条件，匹配失败，由于mid状态存在Ingore状态迁移边，所以当前mid状态不发生变化，event2继续往回匹配start的条件，匹配失败，这时event2被丢弃；同样的event3也不会影响nfa的所有状态，被丢弃。当第五条消息event4来临时，匹配mid的条件成功，更新当前mid状态，并且进入“下一个状态”，那就是endstat状态，说明匹配结束，存在第二个匹配结果\[event1, event4\]。

## 关于cep state的问题 ##

参考文章：[https://bbs.huaweicloud.com/blogs/136695][https_bbs.huaweicloud.com_blogs_136695]

一开始会把所有可以作为CEP匹配头的状态State先放入queue,每来一条数据就会遍历queue中所有state，看这条数据是否能能匹配上，能匹配上就在queue中加入下一个用于匹配的状态，  用于看下一条数据能否继续匹配上

比如一个正则"abc"用于CEP匹配 当来了一条a数据，就匹配上CEP头了，会把b state加入queue中，接着来了一条b数据，又继续匹配上了，又把c state加入queue 直到来了一条c数据整个就匹配完成，返回结果

总结 ： 处理过程就是两步

1.来一条数据，遍历queue中所有state,看哪些state能匹配上就匹配

2.根据1的结果更新queue,用于下一条数据的匹配

**遍历所有完成的状态，当匹配上最后一个状态时就是上面说的“&end”就证明完成了,丢到完成queue中**

**当匹配失败了就清空状态**

**当匹配上了但还没有结束就丢到半匹配queue**

# 总结： #

# 所以当满足第一个条件的时候，cep会缓存state数据，然后NFA会变成take状态，进入下一个条件的匹配，后面的匹配失败会清空状态，如果成功，会丢到完成的队列里面去（然后清空state），，调用用户select方法处理结果，没结束就丢到半匹配队列。所以不太需要考虑state太大太多导致OOM，再加上checkpoint的配合使用，没多大问题。 #

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221022/00bcd0e303a442d8ade11ffb6274c57c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221022/7ac544a0a176406d9544acb8f4737013.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20221022/0f217d267595414785d6ecb07fab12dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20221022/ce1e2758b6ac40c69970890eb86c89d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20221022/b7bd42bdf657473f9c5976b81e8b5a72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5]: /images/20221022/a3369ab99a824263a4a7a14c8360638d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 6]: /images/20221022/3d2de27945604a5fa3adbaa0832c940d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 7]: /images/20221022/b6777561414d4eb78019d73cb58ca278.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 8]: /images/20221022/55db36844cca43eabcfe857ee8dff869.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 9]: /images/20221022/744b954ca9d8459e96cdf5edde65c60b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 10]: /images/20221022/a39bf96a44c3468d9864c0edfadf24e8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 11]: /images/20221022/393fe200ef344c478cbed0eac32ca94d.png
[https_article.itxueyuan.com_PgEMjO]: https://article.itxueyuan.com/PgEMjO
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 12]: /images/20221022/642827d5df614b8ba1e0e46c1389e1ce.png
[preview]: /images/20221022/5438a6c1bf9a4233b2e0e4c97996753e.png
[8ff1c609ccf6722c7af417dfc45b6510.png]: /images/20221022/b8573d1099df42849d8799b7d7eeb14e.png
[https_bbs.huaweicloud.com_blogs_136695]: https://bbs.huaweicloud.com/blogs/136695