flink-cep-paper

发表于 | 分类于 | | |
字数统计 1.4k | 阅读时长 6

CEP实现论文

引言

2016年1月29号，社区提了这个patch[FLINK-3216]引入flink cep模块，flink cep实现基于论文：Efficient Pattern Matching over Event Streams。我们来深入阅读以下这篇论文。

event pattern查询表达形式

先看三个查询

Query1

匹配出的是：挑选从货架取出的商品，未经checkout，带离了商店的行为序列

  1. SEQ(Shelf a, ∼(Register b), Exit c)
  2. where条件指定了a，b，c的tagid相同
  3. within指定窗口大小12小时

Query2

匹配出的是：一个污染报警和相关受污染的发货卸货流转站点序列

  1. 报警类型是污染
  2. 匹配出的装载点和前一个到达点一致
  3. within指定匹配窗口大小是3小时

Query3

匹配出的是：匹配股票交易拐点

  1. 初始成交量高于1000
  2. 持续增加，最后成交量突然跌为最高点的80%以下的序列
  3. 1小时的时间窗口

可以看出规则是怎么描述的：

  1. 指定匹配出的序列是什么（structure）
  2. 限制或筛选条件

还需要匹配策略来进行规制匹配。

  1. 严格的邻近匹配（Strict contiguity）：只有连续两个event能满足匹配，中间不能夹杂其他不符合条件的元素
  2. 分区邻近匹配（Partition contiguity）：这就是说被选出的两个event之间不一定需要完全紧密相连，但是在一个分区中的需要紧密相连 如Query3中的a[]需要连续，a与b不需要连续
  3. skip_till_next_match: 所有的不相关的元素都会被跳过，不再考虑是不是邻近的元素
  4. skip_till_any_match: Query2解释了这种用法。比如当前最后识别的shipment到达了X位置。这时候读进来一条，X->Y的Event，在skip_till_any_match匹配规则下系统会做两件事：1.接收X->Y事件，更新状态2.在当前状态的另一个实例中不接收他来保留一个状态。这样当下一个比如来了个X->Z事件那还是可以接受这个实例。这种策略就是返回所有的可能的序列（allowing non-deterministic actions）在接收到同一个事件后可以有不一样的走向

以上介绍了event pattern的查询形式，接下来我们看其内部实现算法。

NFA 非确定性自动状态机

非确定性自动状态机由以下结构组成。A = (Q,E,θ,q1,F),

Q: 一系列状态
E： 边
θ： 计算公式
q1：起始状态
F: 结束状态

States

the start state,a[1], is where the matching process begins. It awaits input to start the Kleene plus and to select an event into the a[1] unit of the match
buffer. At the next state a[i], it attempts to select another event into the a[i] (i > 1) unit of the buffer. The subsequent state b denotes that the matching process has fulfilled the Kleene plus (for a particular match) and is ready to process the next pattern component. The final state, F, represents
the completion of the process, resulting in the creation of a pattern match.

Edges

Each state is associated with a number of edges,
representing the actions that can be taken at the state. As Figure 2(a) shows, each state that is a singleton state or the first state,p[1], of a pair has a forward begin edge. Each second state, p[i], of a pair has a forward proceed edge, and a looping take edge. Every state (except the start and final states) has a looping ignore edge. The start state has no edges to it as we are only interested in matches that start with selected events.
Each edge at a state, q, is precisely described by a triplet:(1) a formula that specifies the condition on taking it, denoted by θq-edge, (2) an operation on the input stream (i.e.,consume an event or not), and (3) an operation on the match buffer (i.e., write to the buffer or not). Formulas of edges are compiled from pattern queries, which we explain in detail shortly. As shown in Figure 2(a), we use solid lines to denote begin and take edges that consume an event from the input and write it to the buffer, and dashed lines for ignore edges that consume an event but do not write it to the buffer. The proceed edge is a special edge: it does not consume any input event but only evaluates its formula and tries proceeding. We distinguish the proceed edge from ignore edges in the style of arrow, denoting its behavior.

非确定性
某些state具有两条边，但是这两条边的行为不一定是完全相反的

NFA算法过程：

  1. 先根据整个structure构建state和边的行为
  2. 将condition翻译成边上的公式
  3. 将select strategy翻译成边上的公式
  4. 将时间窗口翻译成：最先匹配的元素和最后的元素时长不能超过窗口大小

优化：

  1. 先做过滤
  2. 将window条件前置
  3. 优化proceed边，进入下一个state的时候要检查是否满足begin的条件判断

带版本的内存共享

通过带版本的共享内存解决多runs之间的重复事件问题。

Computation state

  1. version number of a run
  2. current automaton state the run is in
  3. a pointer to the most revent event selected into the buffer
  4. a vector => 保留做edge判断的所需的最少的元素。例如上图中存储了a[i]的和和count为了计算平均值

Merge相同的Runs

将Query3中的avg算法改为max算法，他们在接收到e4之后达到了同样的状态，因此其之后的运算可以merge为一个。

  1. 首先探测什么时候两个runs一样了，引入了算子M。每一个状态机都有一个mask，每一个mask对每一个Vector中列都有一个bit位，都相同时那么这两个run就是相等的。
  2. 创建combined的run
    1. 时间设置为最小的那个时间

Backtrack算法（回溯）

每次只跑一个run，跑失败了再回退到分叉点跑另一条。也就是以广度优先搜索方式 breadth first search manner

待补充

  1. 性能评估方案
  2. 有哪些因素会影响计算速度和吞吐
  3. 内存优化管理措施
谢谢支持