1.什么是 CEP

概念：

　　一个或多个由简单事件构成的事件流通过一定的规则匹配，然后输出用户想得到的数据，满足规则的复杂事件。

特征：

 目标：从有序的简单事件流中发现一些高阶特征

 输入：一个或多个由简单事件构成的事件流

 处理：识别简单事件之间的内在联系，多个符合一定规则的简单事件构成复杂事件

 输出：满足规则的复杂事件

　　CEP 用于分析低延迟、频繁产生的不同来源的事件流。CEP 可以帮助在复杂的、不相关的事件流中找出有意义的模式和复杂的关系，以接近实时或准实时的获得通知并阻止一些行为。

　　CEP 支持在流上进行模式匹配，根据模式的条件不同，分为连续的条件或不连续的条件；模式的条件允许有时间的限制，当在条件范围内没有达到满足的条件时，

会导致模式匹配超时。

　　看起来很简单，但是它有很多不同的功能：

　　 输入的流数据，尽快产生结果

　　 在 2 个 event 流上，基于时间进行聚合类的计算

　　 提供实时/准实时的警告和通知

　　 在多样的数据源中产生关联并分析模式

　　 高吞吐、低延迟的处理

　　市场上有多种 CEP 的解决方案，例如 Spark、Samza、Beam 等，但他们都没有提供专门的 library 支持。但是 Flink 提供了专门的 CEP library。

2 Flink CEP

Flink 为 CEP 提供了专门的 Flink CEP library，它包含如下组件：

 Event Stream

 pattern 定义

 pattern 检测

 生成 Alert

　　首先，开发人员要在 DataStream 流上定义出模式条件，之后 Flink CEP 引擎进行模式检测，必要时生成告警。

　　为了使用 Flink CEP，我们需要导入依赖：

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>flink-cep_${scala.binary.version}</artifactId>
 <version>${flink.version}</version>
</dependency>

Event Streams

以登陆事件流为例：

case class LoginEvent(userId: String, ip: String, eventType: String, eventTime: String)

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.setParallelism(1)

val loginEventStream = env.fromCollection(List(
 LoginEvent("1", "192.168.0.1", "fail", "1558430842"),
 LoginEvent("1", "192.168.0.2", "fail", "1558430843"),
 LoginEvent("1", "192.168.0.3", "fail", "1558430844"),
 LoginEvent("2", "192.168.10.10", "success", "1558430845")
)).assignAscendingTimestamps(_.eventTime.toLong)

3 Pattern API

4 个体模式

个体模式的条件

5 模式序列

6 模式检测

7 匹配事件提取

8 超时事件的提取