Flink学习笔记(10) - CEP

2022-09-01 16:45:22 浏览数 (1)

一、什么是CEP?

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

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

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

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

二、CEP的特点

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

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

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

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

三、Pattern API

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

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

个体模式(Individual Patterns)

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

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

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

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

模式组(Groups of patterns)

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

四、个体模式(Individual Patterns)

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

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

量词(Quantifier)

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

条件(Condition)

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

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

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

简单条件(Simple Condition)

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

组合条件(Combining Condition)

  将简单条件进行合并;.or() 方法表示或逻辑相连,where 的直接组合就是 AND

终止条件(Stop Condition)

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

迭代条件(Iterative Condition)

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

  调用 .where( (value, ctx) => {…} ),可以调用 ctx.getEventsForPattern(“name”)

五、模式序列

不同的“近邻”模式   1 严格近邻

  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() —— 不想让某个事件在两个事件之间发生

需要注意:

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

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

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

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

六、模式的检测

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

  调用 CEP.pattern(),给定输入流和模式,就能得到一个 PatternStream

七、匹配事件的提取

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

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

  select() 以一个 Map[String,Iterable [IN]] 来接收匹配到的事件序列,其中 key 就是每个模式的名称,而 value 就是所有接收到的事件的 Iterable 类型

八、超时事件的提取

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

  超时处理程序会接收到目前为止由模式匹配到的所有事件,由一个 OutputTag 定义接收到的超时事件序列

0 人点赞