FlinkCEP 中提供了 Pattern API 用于对输入流数据的复杂事件规则定义，并从事件流 中抽取事件结果。包含四个步骤：

输入事件流的创建
Pattern 的定义
Pattern 应用在事件流上检测
选取结果

一.模式定义

定义 Pattern 可以是单次执行模式，也可以是循环执行模式。单词执行模式一次只接受 一个事件，循环执行模式可以接收一个或者多个事件。通常情况下，可以通过指定循环次数 将单次执行模式变为循环执行模式。每种模式能够将多个条件组合应用到同一事件之上，条 件组合可以通过 where 方法进行叠加。每个 Pattern 都是通过 begin 方法定义的

val start = Pattern.begin[Event]("start_pattern")

下一步通过 Pattern.where()方法在 Pattern 上指定 Condition，只有当 Condition 满 足之后，当前的 Pattern 才会接受事件。

start.where(_.getCallType == "success")

1. 设置循环次数
   对于已经创建好的 Pattern，可以指定循环次数，形成循环执行的 Pattern。
   times：可以通过 times 指定固定的循环执行次数。

//指定循环触发4次 
start.times(4); 
//可以执行触发次数范围,让循环执行次数在该范围之内 
start.times(2, 4);

2. optional：也可以通过 optional 关键字指定要么不触发要么触发指定的次数。

start.times(4).optional(); 
start.times(2, 4).optional();

3. greedy：可以通过 greedy 将 Pattern 标记为贪婪模式，在 Pattern 匹配成功的前提下， 会尽可能多地触发。

//触发2、3、4次,尽可能重复执行 
start.times(2, 4).greedy(); 
//触发0、2、3、4次,尽可能重复执行 
start.times(2, 4).optional().greedy();

4. oneOrMore：可以通过 oneOrMore 方法指定触发一次或多次。

// 触发一次或者多次 
start.oneOrMore(); 
//触发一次或者多次,尽可能重复执行 
start.oneOrMore().greedy(); 
// 触发0次或者多次 
start.oneOrMore().optional(); 
// 触发0次或者多次,尽可能重复执行 
start.oneOrMore().optional().greedy();

5. timesOrMor：通过 timesOrMore 方法可以指定触发固定次数以上，例如执行两次以上。

// 触发两次或者多次 
start.timesOrMore(2); 
// 触发两次或者多次,尽可能重复执行 




    

start.timesOrMore(2).greedy(); 
// 不触发或者触发两次以上,尽可能重复执行 
start.timesOrMore(2).optional().greedy();

1.1.定义条件

每个模式都需要指定触发条件，作为事件进入到该模式是否接受的判断依据，当事件中 的数值满足了条件时，便进行下一步操作。在 FlinkCFP 中通过 pattern.where()、 pattern.or()及 pattern.until()方法来为 Pattern 指定条件，且 Pattern 条件有 Simple Conditions 及 Combining Conditions 等类型。

1. 简单条件：Simple Condition 继承于 Iterative Condition 类，其主要根据事件中的 字段信息进行判断，决定是否接受该事件。

// 把通话成功的事件挑选出来 
start.where(_.getCallType == "success")

2. 组合条件：组合条件是将简单条件进行合并，通常情况下也可以使用 where 方法进行条 件的组合，默认每个条件通过 AND 逻辑相连。如果需要使用 OR 逻辑，直接使用 or 方法 连接条件即可。

// 把通话成功，或者通话时长大于10秒的事件挑选出来 
val start = Pattern.begin[StationLog]("start_pattern") 
.where(_.callType=="success") 
.or(_.duration>10)

3. 终止条件：如果程序中使用了 oneOrMore 或者 oneOrMore().optional()方法，则必须 指定终止条件，否则模式中的规则会一直循环下去，如下终止条件通过 until()方法指 定

pattern.oneOrMore.until(_.callOut.startsWith("186"))

1.2.模式序列

将相互独立的模式进行组合然后形成模式序列。模式序列基本的编写方式和独立模式一致，各个模式之间通过邻近条件进行连接即可，其中有严格邻近、宽松邻近、非确定宽松邻 近三种邻近连接条件。

1. 严格邻近：严格邻近条件中，需要所有的事件都按照顺序满足模式条件，不允许忽略任 意不满足的模式

val strict: Pattern[Event] = start.next("middle").where(...)

2. 宽松邻近：在宽松邻近条件下，会忽略没有成功匹配模式条件，并不会像严格邻近要求 得那么高，可以简单理解为 OR 的逻辑关系。

val relaxed: Pattern[Event, _] = start.followedBy("middle").where(...)

3. 非确定宽松邻近：和宽松邻近条件相比，非确定宽松邻近条件指在模式匹配过程中可以 忽略已经匹配的条件。

val nonDetermin: Pattern[Event, _] = start.followedByAny("middle").where(...)

4. 除以上模式序列外，还可以定义“不希望出现某种近邻关系”：
   notNext() —— 不想让某个事件严格紧邻前一个事件发生
   notFollowedBy() —— 不想让某个事件在两个事件之间发生

注意：
1、所有模式序列必须以 .begin() 开始
2、模式序列不能以 .notFollowedBy() 结束
3、“not” 类型的模式不能被 optional 所修饰
4、此外，还可以为模式指定时间约束，用来要求在多长时间内匹配有效

二.模式检测

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

//cep 做模式检测 
val patternStream = CEP.pattern[EventLog](dataStream.keyBy(_.id),pattern)

三.选择结果

得到 PatternStream 类型的数据集后，接下来数据获取都基于 PatternStream 进行。该 数据集中包含了所有的匹配事件。目前在 FlinkCEP 中提供 select 和 flatSelect 两种方法 从 PatternStream 提取事件结果事件。

1. 通过 Select Funciton 抽取正常事件
   可以通过在 PatternStream 的 Select 方法中传入自定义 Select Funciton 完成对匹配 事件的转换与输出。其中 Select Funciton 的输入参数为 Map[String, Iterable[IN]]，Map 中的 key 为模式序列中的 Pattern 名称，Value 为对应 Pattern 所接受的事件集合，格式为 输入事件的数据类型。
2. 通过 Flat Select Funciton 抽取正常事件 Flat Select Funciton 和 Select Function 相似，不过 Flat Select Funciton 在每次 调用可以返回任意数量的结果。因为 Flat Select Funciton 使用 Collector 作为返回结果 的容器，可以将需要输出的事件都放置在 Collector 中返回。
3. 通过 Select Funciton 抽取超时事件 如果模式中有 within(time)，那么就很有可能有超时的数据存在，通过 PatternStream. Select 方法分别获取超时事件和正常事件。首先需要创建 OutputTag 来标记超时事件，然 后在 PatternStream.select 方法中使用 OutputTag，就可以将超时事件从 PatternStream 中抽取出来。