Flink-CEP复杂事件处理

淡淡的烟草味﹌ 2022-09-11 00:26 169阅读 0赞

# CEP #

复杂事件处理(Complex Event Processing,CEP)

Flink CEP是在 Flink 中实现的复杂事件处理(CEP)库

CEP 允许在无休止的事件流中检测事件模式,让我们有机会掌握数据中重要的部分

一个或多个由简单事件构成的事件流通过一定的规则匹配,然后输出用户想得到的数据 —— 满足规则的复杂事件

## CEP 的特点 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]

• 目标:从有序的简单事件流中发现一些高阶特征

• 输入:一个或多个由简单事件构成的事件流

• 处理:识别简单事件之间的内在联系,多个符合一定规则的简单事件构成复杂事件

• 输出:满足规则的复杂事件

## Pattern API ##

处理事件的规则,被叫做“模式”(Pattern)

Flink CEP 提供了 Pattern API,用于对输入流数据进行复杂事件规则定义,用来提取符合规则的事件序列

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]

**个体模式(Individual Patterns)**

– 组成复杂规则的每一个单独的模式定义,就是“个体模式”

![在这里插入图片描述][ea1c07e214514a97950070cfb165ec9e.png]

**组合模式(Combining Patterns,也叫模式序列)**

– 很多个体模式组合起来,就形成了整个的模式序列

– 模式序列必须以一个“初始模式”开始:

![在这里插入图片描述][65e51c20749249908eacff0002cb9bb3.png]

**模式组(Groups of patterns)**

– 将一个模式序列作为条件嵌套在个体模式里,成为一组模式

### 个体模式(Individual Patterns) ###

个体模式可以包括“单例(singleton)模式”和“循环(looping)模式”

单例模式只接收一个事件,而循环模式可以接收多个

**量词(Quantifier)**

– 可以在一个个体模式后追加量词,也就是指定循环次数

![在这里插入图片描述][337b363dfef749579dda0c6f938aea2d.png]

**条件(Condition)**

– 每个模式都需要指定触发条件,作为模式是否接受事件进入的判断依据

– CEP 中的个体模式主要通过调用 .where() .or() 和 .until() 来指定条件

– 按不同的调用方式,可以分成以下几类:

**简单条件(Simple Condition)**

– 通过 .where() 方法对事件中的字段进行判断筛选,决定是否接受该事件

![在这里插入图片描述][4aa961bbea7f4413ac8ffff482c91db9.png]

**组合条件(Combining Condition)**

– 将简单条件进行合并;.or() 方法表示或逻辑相连,where 的直接组合就是 AND（多个where串联）  
![在这里插入图片描述][b7286ec3b79444f39fffd23cd3e382b3.png]

**终止条件(Stop Condition)**

– 如果使用了 oneOrMore 或者 oneOrMore.optional,建议使用 .until() 作为终止条件,以便清理状态

**迭代条件(Iterative Condition)**

– 能够对模式之前所有接收的事件进行处理

– 可以调用 ctx.getEventsForPattern(“name”)，可以获取到上下文

![在这里插入图片描述][b1806ce73e6246c58a1fc7542f8ed8a4.png]

## 模式序列 ##

**不同的“近邻”模式**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 2]

**严格近邻(Strict Contiguity)**

– 所有事件按照严格的顺序出现,中间没有任何不匹配的事件,由 .next() 指定

– 例如对于模式”a next b”,事件序列 \[a, c, b1, b2\] 没有匹配

**宽松近邻( Relaxed Contiguity )**

– 允许中间出现不匹配的事件,由 .followedBy() 指定

– 例如对于模式”a followedBy b”,事件序列 \[a, c, b1, b2\] 匹配为 \{a, b1\}

**非确定性宽松近邻( Non-Deterministic Relaxed Contiguity )**

– 进一步放宽条件,之前已经匹配过的事件也可以再次使用,由 .followedByAny() 指定

– 例如对于模式”a followedByAny b”,事件序列 \[a, c, b1, b2\] 匹配为 \{a, b1\},\{a,b2\}

**除以上模式序列外,还可以定义“不希望出现某种近邻关系”:**

– .notNext() —— 不想让某个事件严格紧邻前一个事件发生

– .notFollowedBy() —— 不想让某个事件在两个事件之间发生

**需要注意:**

– 所有模式序列必须以 .begin() 开始

– 模式序列不能以 .notFollowedBy() 结束

– “not” 类型的模式不能被 optional 所修饰

– 此外,还可以为模式指定时间约束,用来要求在多长时间内匹配有效

![在这里插入图片描述][ac7e1adfb575488db5b1a90609296a43.png]

## 模式的检测 ##

指定要查找的模式序列后,就可以将其应用于输入流以检测潜在匹配

调用 CEP.pattern(),给定输入流和模式,就能得到一个PatternStream  
![在这里插入图片描述][293719da7fcd43b09b87ac6caa406ef4.png]

## 匹配事件的提取 ##

创建 PatternStream 之后,就可以应用 select 或者 flatselect 方法,从检测到的事件序列中提取事件了

select() 方法需要输入一个 select function 作为参数,每个成功匹配的事件序列都会调用它

select() 以一个 Map<String,List > 来接收匹配到的事件序列,其中 key就是每个模式的名称,而 value 就是所有接收到的事件的 List 类型  
![在这里插入图片描述][da224c8b60624fd8b2d69ce4d696b191.png]

## 超时事件的提取 ##

当一个模式通过 within 关键字定义了检测窗口时间时,部分事件序列可能因为超过窗口长度而被丢弃;为了能够处理这些超时的部分匹配,select 和 flatSelect API 调用允许指定超时处理程序

超时处理程序会接收到目前为止由模式匹配到的所有事件,由一个 OutputTag 定义接收到的超时事件序列  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 3]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/f8d0e87550e94c15a4e3aafbf1e26973.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/96ec2d8ab1324df2bd9c15b5047c7dbf.png
[ea1c07e214514a97950070cfb165ec9e.png]: /images/20220829/7de6cb3c87154f848c0105e40135e2a6.png
[65e51c20749249908eacff0002cb9bb3.png]: /images/20220829/2a4892f0e67848dd8b2b444baa35567f.png
[337b363dfef749579dda0c6f938aea2d.png]: /images/20220829/23f40a111fd54e25a82f8c023ba55ec1.png
[4aa961bbea7f4413ac8ffff482c91db9.png]: /images/20220829/9fff62ede14e4a23a77d103409d44505.png
[b7286ec3b79444f39fffd23cd3e382b3.png]: /images/20220829/9f3c85a063f7487aa1e95906a4aa874a.png
[b1806ce73e6246c58a1fc7542f8ed8a4.png]: /images/20220829/ae6412d73e4b4f44ae6c335937831a73.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/c058a0935dfb4e35803a54e490449535.png
[ac7e1adfb575488db5b1a90609296a43.png]: /images/20220829/05784f2af47349778eae7e88c0baff5c.png
[293719da7fcd43b09b87ac6caa406ef4.png]: /images/20220829/f62fc2d7977e47558601164a76141488.png
[da224c8b60624fd8b2d69ce4d696b191.png]: /images/20220829/8968b411d2714f389c020ff0bde852cf.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/6a169248cbc44291a5bcf0da5aab4a06.png