和美大家说 | 基于Flink CEP的实时页面时长统计

Flink CEP 也叫做Flink复杂事件处理，可以在无穷无界的事件流中检测事件规则，通过模式规则匹配的方式对重要信息进行跟踪和分析，从而在实时数据中发掘出有价值的信息。

复杂事件处理主要应用于防范网络欺诈、设备故障检测、风险规避和智能营销等领域。

Flink 基于 DataStrem API 提供了 FlinkCEP 组件栈，专门用于对复杂事件的处理，帮助用户从流式数据中发掘出有价值的信息。

FlinkCEP 中提供了 Pattern API 用于对输入流数据的复杂事件规则定义，并从事件流中抽取事件结果。其中涉及三个部分：

对于已经创建好的 Pattern，可以指定循环次数，形成循环执行的 Pattern链。

🔷 times：可以通过 times 指定固定的循环执行次数。 

//指定循环触发4次

start.times(4);

//可以执行触发次数范围,让循环执行次数在该范围之内

start.times(2, 4); 

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

start.times(4).optional();

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

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

//触发2、3、4次,尽可能重复执行

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

//触发0、2、3、4次,尽可能重复执行

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

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

// 触发一次或者多次

start.oneOrMore();

//触发一次或者多次,尽可能重复执行

start.oneOrMore().greedy();

// 触发0次或者多次

start.oneOrMore().optional();

// 触发0次或者多次,尽可能重复执行

start.oneOrMore().optional().greedy(); 

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

// 触发两次或者多次

start.timesOrMore(2);

// 触发两次或者多次,尽可能重复执行

start.timesOrMore(2).greedy();

// 不触发或者触发两次以上,尽可能重复执行

start.timesOrMore(2).optional().greedy(); 

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

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

// 把通话成功的事件挑选出来

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

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

// 把通话成功，或者通话时长大于10秒的事件挑选出来

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

.where(_.callType=="success")

.or(_.duration>10) 

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

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

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

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

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

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

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

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

//指定模式在10秒内有效

pattern.within(Time.seconds(10)); 

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

//cep 做模式检测

val patternStream=CEP.pattern[EventLog(dataStream.keyBy(_.id),pattern)

得到 PatternStream 类型的数据集后，接下来数据获取都基于 PatternStream 进行。该数据集中包含了所有的匹配事件。

目前在 FlinkCEP 中提供 select 和 flatSelect 两种方法从 PatternStream 提取事件结果事件。 

通过Flink CEP匹配业务开始和结束标记，实时计算业务的耗时情况，用于确认业务办理是否超时，从而提醒现场人员进行实时干预。

其程序流程图如下：

部分代码如下：