flink1.12.x FlinkCEP-模式：.where()、begin()、next()、followedBy()

作者：马育民 • • 阅读：10206

介绍

处理事件的规则，被叫作模式（Pattern）。

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

模式大致分为三类：

  • 个体模式（Individual Patterns）：组成复杂规则的每一个单独的模式定义，就是个体模式。

  • 组合模式（Combining Patterns，也叫 模式序列）：很多个体模式组合起来，就形成了整个的模式序列。
    模式序列必须以一个初始模式开始：

  • 模式组（Group of Pattern）：将一个模式序列作为条件嵌套在个体模式里，成为一组模式。

个体模式

个体模式一般都会匹配接收 一个事件

个体模式包括 单例模式 和 循环模式：

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

条件

每个模式都需要指定触发条件，作为模式是否接受事件进入的判断依据。CEP中的个体模式主要通过调用 .where()、.or() 和 .until() 来指定条件。

按不同的调用方式，可以分成以下几类：

  • 简单条件
  • 组合条件
  • 终止条件
  • 迭代条件
  • 限定匹配事件的类型

简单条件

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

pattern.where(new SimpleCondition<Event>() { 
    @Override
    public boolean filter(Event value) {
        return value.user.startsWith("A");
    }
});

组合条件

当个体模式 有多个限定条件时，有下面解决方法：

  • 在 .filter() 方法中，通过 if-else 的条件分支分别定义多个条件。缺点：代码臃肿，可读性降低

  • 独立定义 多个条件，然后 连接起来，构成一个 组合条件 。or() 方法表示逻辑或，where 的直接组合就相当于与 and

例子：

pattern.where(new SimpleCondition<Event>() {
    @Override
    public boolean filter(Event value) {
        return value.user.startsWith("A");
    }
})
.or(new SimpleCondition<Event>() {
    @Override
    public boolean filter(Event value) {
        return value.user.startsWith("A");
    }
});

终止条件

终止条件只与 oneOrMore() 或者 oneOrMore().optional() 结合使用。

在这种循环模式下，不知道后面 还有没有事件可以匹配，只好把之前匹配的事件作为状态缓存起来继续等待，这等待无穷无尽；如果一直等下去， 缓存的状态越来越多，最终会耗尽内存。

所以这种循环模式必须有个终点，当 .until() 指定的条件满足时，循环终止，这样就可以清空状态释放内存了。

组合模式（模式序列）

类似处理 连续登录失败3次，就需要用到 组合模式（模式序列）

将多个 个体模式，按一定的顺序 连接起来，定义一个完整的复杂事件匹配规则，就叫作 组合模式（Combining Pattern），也叫作 模式序列（Pattern Sequence）。

代码套路

Pattern<Event, ?> pattern = Pattern
    .<Event>begin("start").where(...)
    .next("next").where(...)
    .followedBy("follow").where(...)
    ...

用 begin()、next()、followedBy() 等表示先后顺序的 方法 将 个体模式串连起来，每个事件 匹配的条件是什么、各个事件之间谁先谁后、近邻关系如何 都定义得一目了然。

这些方法调用之后，返回的仍然是一个 Pattern 的对象

初始模式（Initial Pattern）

所有的组合模式，都必须以一个 初始模式 开头，如下：

Pattern<Event, ?> start = Pattern.<Event>begin( name );

形参：

  • name：模式的名称；
  • 泛型 Event：需要传入一个类型参数，模式要检测流中事件的基本类型。

返回值： 调用的结果返回一个 Pattern 对象，有两个泛型参数：

  1. 第一个就是检测事件的基本类型 Event，跟 begin 指定的类型一致
  2. 第二个则是当前模式里事件的子类型，由子类型限制条件指定。这里用类型通配符（？）代替，就可以从上下文推断

近邻条件（Contiguity Conditions）

模式之间的组合是通过一些 begin()、next()、followedBy() 方法实现的，这些方法指明了 先后事件 之间有着怎样的 近邻关系，这就是所谓的 近邻条件（Contiguity Conditions，也叫 连续性条件）。

提供了三种近邻关系：

  • 严格近邻（Strict Contiguity）
  • 非确定性宽松近邻（Non-Deterministic Relaxed Contiguity）

严格近邻（Strict Contiguity）

如图所示，匹配的事件严格 地按顺序一个接一个出现，中间 不会有任何其他事件。

代码中对应的就是 Pattern 的 .next() 方法，是紧挨着的

宽松近邻

如图所示，宽松近邻只关心 事件发生的顺序，两个匹配的事件 之间 可以有其他不匹配的事件出现

代码中对应 Pattern 的 .followedBy()方法，表示 跟在后面 就可以，不需要紧紧相邻。

非确定性宽松近邻

这种近邻关系更加宽松。所谓 非确定性 是指：可以 重复使用 之前已经 匹配 过的事件

这种近邻条件下匹配到的不同复杂事件，可以以同一个事件作为开始，所以匹配结果一般会比宽松近邻更多，如图所示。

代码中对应 Pattern 的 .followedByAny()方法。

参考：
参考：
https://blog.csdn.net/mengxianglong123/article/details/124123750
https://cloud.tencent.com/developer/article/1683066

原文出处：/show_1IX4GGASFcaS.html

吉ICP备15003043号