flink cep 主要用来处理复杂事件的库，如实时提取一段时间内股票涨，跌，涨，跌的事件并作出对应的决策。
官网地址https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/libs/cep.html

需求1:要在事件序列中提取 连续数据是 10，20，10 的事件
需求2:要在事件序列中提取 连续数据是 10 (10-50)+ 0 （第一个数是10，后续连续到来一个或者多个10到50之间的数字(终止条件：出现大于50的数字)）的事件 最后到来一个是0的数字
如下代码实现

如上需求和代码 其中重点就是创建匹配模式具体的api可以官网上面详细查看，接下来我们探究下如何提取复杂事件模型

以下内容参考 《基于复杂事件处理的模型和算法研究》-魏仕杰
一，查询的约束条件（sase+语言为例）
查询约束条件就是确定事件的约束条件
查询的约束条件分为两类：时序约束和谓词约束
时序约束：描述的是事件间的时序关系的约束
谓词约束：描述的是单个事件的属性的约束
时序约束分类：
1，序列算子：定义了模式匹配结构为按照时间先后顺序得到事件序列的顺序结构，用SEQ表示，在PATTERN子句中定义，例如定义SEQ(A,B,C) 事件类型 A,B,C 按照时间顺序A->B->C的方式匹配 在flink中类似 begin, next,follewby 方法对状态进行组合
2，时间窗口：规定一个时间，约束事件在这段时间内发生 在within 语句中定义。在flunk中和within 类似
3，克林闭包算子：表示一个事件发生一次或者多次的事件序列，如A+表示事件A发生一次或者多次的事件序列，类似flunk当中的oneormore()方法
4，扩展克林算子：对克林闭包的扩展，A[n,m]表示事件A发生n到m次的事件序列，类似flunk当中的time函数
谓词约束：是针对事件属性的约束，可以是单个事件的属性约束（如订单金额大于100）
也可以是事件间的属性约束（如当前订单金额大于前一笔订单金额）
类似flunk当中的where方法
二，NFA～b模型
NFA～b是非确定性有限自动机(NFA)结合缓存匹配的模型。是NFA模型的改进，其中匹配缓存用与在运行时存储查询出来的结果。
如下图示：

NFA～b模型由三个部分组成
1，状态
在非确定有限自动机中状态有两种情况：
a）非克林闭包事件由一个状态来表示
b）克林闭包事件由两个状态来表示
上图中每个圆圈代表一个状态其中 A事件分为两个A[1] 和A[i]。A[1]是该克林闭包算子的开始条件，当事件到达满足A[1]时候表示A事件开始(开始克林闭包算子)，并且把A[1]匹配的事件放入A[1]状态对应的缓存当中，到达下一个状态A[i]，当下一个事件到达和A[i]匹配的则放入A[i]所对应的缓存当中。当克林算子满足终止条件终止时候(例如A[] 只匹配十次 和flink的time(10)方法类似),表示匹配过程已经满足克林闭包查询条件约束，接着进入到下一个状态B，B是一个非克林闭包事件的状态，当B匹配成功同样把事件存入B的缓存当中进入到最终状态F，表示匹配结束
2，有向边
与状态有关的有向边代表事件到达该状态时可以采取的操作。执行操作的条件由模型指定的事件类型，约束条件，终止条件，事件选择策略等决定。
每条有向边代表着状态越迁的条件，对事件流的操作(是否消耗事件)，是否把事件存入缓存之中，其中图中的实线表示消耗事件并且把事件存入缓存如begin take ,图中ignore对应的虚线表示消耗事件但是不存入缓存，即忽略这个事件，图中process对应的虚线表示不消耗事件也不存入缓存，此时事件进入下一个状态
有向边分为四种：
begin：消费事件存入缓存状态越迁到下一个状态
ignore：表示跳过不相关事件并且依然处于该状态被忽略事件不存入缓存，开始状态没有ignore边因为匹配必须从一个选定的事件开始，事件是否忽略和有向边的执行条件有关(下文讲解)
take：表示选择当前时间，消费事件并且存入缓存中，还是处于当前状态，是否take和有向边的执行条件有关(下文讲解)。
process：表示完成克林闭包的选择策略，如设置啦次数限制等终止条件或者时间窗口的限制，状态越迁到下一个状态 是否process和有向边的执行条件有关（下文讲解）
3，run（运行）
模型会产生多个可能的事件序列也就是多个run，每个run表示对到达事件的一次处理，处理结果存入缓存当中。一个run代表一个根据模型匹配的事件序列。一个事件到来会对多个run进行匹配。

NFA~b的条件设定
如上图中的 A[1]_begin ，表示begin有向边的条件(a[1].price=20)，其他有向边也有对应的条件，有向边的执行条件有如下：
1，相关事件定义：在where语句中定义的谓词约束即位相关事件，如A[1]_begin 中的谓词约束 a[1].price=20，这些谓词约束用来设置begin和take执行的条件，可以是单个事件的属性值约束也可以是多个事件之间的关系约束，或者是对事件聚合后的结果约束。不满足约束的是无关事件，满足的即为相关事件。
2，事件选择策略：从事件流中选择和克林闭包算子相关的事件时，需要有相关的选择策略，如一些查询希望在事件流中查询连续的事件，而另外的希望查询不连续的事件。
skip till next match：在策略中忽略无关事件，对相邻的相关事件进行连续的匹配选择，注意是连续的相关事件。两个相关事件之间的相关事件都的选择。克林闭包使用该策略直到满足终止原则的条件，如上query1的查询选择从price从20开始并且不断增长的事件，忽略掉波动下降的事件。即选择连续的相关事件。
skip till any match：该策略选择的时候更加灵活，考虑如下情况
如果来的事件价格如下 20，30，100，40 ，50 根据上面的选择策略只能选择 20，30，100 的最长序列，但是如果100 是个特例呢，那么后面的40 ，50就丢失啦。此时我们可以略掉100的价格，则事件序列是20，30，40，50。即选择相关事件但是可以不连续，中间可以选择的忽略掉一些相关事件。对每一个到来的相关事件都可以对他作出非确定的决策：选择加入还是忽略跳过。
3，终止准则：克林闭包不能无限匹配选择下去，得设置一个终止准则，如上图中的a[i]_foced-stop = a[i].time-a[1].time>w || b_begin (到达设置的时间窗口事件，或者b事件开始)

以下内容参考论文：https://people.cs.umass.edu/~yanlei/publications/sase-sigmod08.pdf

一，Shared Versioned Match Buffer
共享的版本的匹配缓存：一个存储各个state中匹配的事件和事件间关系(带有版本)的共享的缓存
思考：如果到来一个事件可以take 也可以process 即由一个run分裂为两个run，那么这两个run的事件序列该如何存储呢？
实例：我们以以下实例进行分析
1，查询语句
查询描述：以大于1000股票交易量的开始 之后当价格持续增长时候 交易量大跌

2，事件序列：

如上序列：在状态为a[i]时，e6到来，此时750<999*0.8 满足 b的take条件，同时125>(100+120+120+121+120)/5 满足a[i]的take条件，此时一个run会分裂为两个如上的 R1和R3，此时两个run当中前五个事件是相同的，如果分别为每个run都建立一个缓存会有大量的冗余，而且每个事件都可能产生分裂，所以冗余量可能会很大下图中的三个run建立三个buffer a,b,c 数据会有大量重复，而d中带上版本号的数据就会减少冗余。

3，缓存模型

如上图所示，用右边的带有版本号的边去关联数据，来代替左边的三个buffer。
我们如何选出 e3,e4,e6 的事件序列 或e1,e2,e3,e4,e5,e6 的事件序列呢？
如果不带版本号的话 会出现回溯到e1,e2,e3,e4,e6 的事件序列，因为e6->e4->e3 中 e4 ->e3 存在两条边，而我们只需要 e6->e4->e3 的序列 。
解决方法：添加版本号，论文中的版本号规则如下
如上 y1.(yi)* 其中i代表当前的state， 如a[i] 中的事件 边的版本只有两个数，b中的版本有三个。yi的值是出现分裂就加一初始值为0 e4<-e5 版本为1.0 e5<-e6 分裂为两条 则在当前state中的e5<-e6 版本加一为 1.1 而到b state的 e5<-e6 数加一为 1.0.0。
那么我们如何通过生成的版本号进行回溯出来最终需要的事件序列呢？
解决方法：回溯时候长度相同的版本号只能最后一位数字不同且是累加的，长度不同的版本号短的版本号必须作为长的版本号的前缀。
如上图中 e4<-e6 的版本号是2.0.0 e3<-e4 有两个版本号，一个2.0 一个是1.0 显然1.0不是2.0.0的前缀所以被排除掉 选择2.0版本的 e3<-e4。最终得到 e3<-e4<-e6 序列 而不是 e1<-e2<-e3<-e4<-e6
flink中版本会有不同 是根据事件数来决定i的值（后文我们通过源码分析）yi的值是当前事件可以产生分裂边的数量决定

二，Computation state
计算状态，每个run会有一个当前的需要进行计算的计算状态，里面保存啦当前run的版本信息当前的所在的state 和一些需要和新到达事件进行比较以判断是否take ignore process的历史事件数据，每次新到达事件会根据Computation state 来进行判断是否take或 ignore或 process。
论文中的数据结构如下
保存啦版本好，当前状态，最近事件，开始事件事件 和一些聚合数据 如 ：总金额总数量来计算平均金额来判断新到达事件是否满足a[i]的条件，还有交易量来判断新到达事件是否满足 b状态条件。当然还有数据结构的优化策略，这里不做介绍。
flink中Computation state会略有不同，下篇文章我们通过源码分析。

参考：
1《基于复杂事件处理的模型和算法研究》-魏仕杰
2 https://people.cs.umass.edu/~yanlei/publications/sase-sigmod08.pdf