什么你还不会Flink的CEP，那你需要好好看看这篇文章

蔚落 2023-02-09 11:52 33阅读 0赞

# Flink之CEP详解 #

# 一、是什么 #

维基百科对CEP的定义为:“CEP是一种事件处理模式，它从若干源中获取事件，并侦测复杂环境的事件或模式，CEP的目的是确认一些有意义的事件（比如某种威胁或某种机会），并尽快对其作出响应”。总结一下也就是CEP是一个事件处理模式，当某项检测需要在多源且复杂的事件流中进行处理，并需要低延迟、秒级或毫秒级的响应时，我们就可以考虑用到它。市场上有多种CEP的解决方案，例如Spark、Samza、Beam等，但他们都没有提供专门的library支持。但是Flink提供了专门的CEP library。

# 二、 Flink CEP详解 #

Flink中实现一个CEP可以总结为四步：

*  构建需要的数据流
 *  构造正确的模式
 *  将数据流和模式进行结合
 *  在模式流中获取匹配到的数据

其中第一步和第三步一般会是标准操作，核心在于第二部构建模式，需要利用Flink CEP支持的特性，构造出正确反映业务需求的匹配模式。

## 2.1 详解Flink CEP library ##

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>

### 2.1.1 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("001", "192.168.0.101", "fail", "1558430842"),
      LoginEvent("001", "192.168.0.102", "fail", "1558430843"),
      LoginEvent("001", "192.168.0.103", "fail", "1558430844"),
      LoginEvent("002", "192.168.10.104", "success", "1558430845")
    )).assignAscendingTimestamps(_.eventTime.toLong)

### 2.1.2 Pattern API ###

每个Pattern都应该包含几个步骤，或者叫做state。从一个state到另一个state，通常我们需要定义一些条件，例如下列的代码：

val loginFailPattern = Pattern.begin[LoginEvent]("begin")
      .where(_.eventType.equals("fail"))
      .next("next")
      .where(_.eventType.equals("fail"))
      .within(Time.seconds(10)

每个state都应该有一个标示：例如`.begin[LoginEvent]("begin")`中的"begin"  
每个state都需要有一个唯一的名字，而且需要一个filter来过滤条件，这个过滤条件定义事件需要符合的条件，例如:  
`.where(_.eventType.equals("fail"))`  
我们也可以通过subtype来限制event的子类型：  
`start.subtype(SubEvent.class).where(...);`  
事实上，你可以多次调用subtype和where方法；而且如果where条件是不相关的，你可以通过or来指定一个单独的filter函数：  
`pattern.where(...).or(...);`  
之后，我们可以在此条件基础上，通过next或者followedBy方法切换到下一个state，next的意思是说上一步符合条件的元素之后紧挨着的元素；而followedBy并不要求一定是挨着的元素。这两者分别称为严格近邻和非严格近邻。

val strictNext = start.next("middle")
    val nonStrictNext = start.followedBy("middle")

最后，我们可以将所有的Pattern的条件限定在一定的时间范围内：  
`next.within(Time.seconds(10))`  
这个时间可以是Processing Time，也可以是Event Time。

注：有人可能会说API介绍的不够详细，所以这里推荐一篇[关于Pattern API详解的博客][Pattern API]

### 2.1.3 Pattern 检测 ###

通过一个input DataStream以及刚刚我们定义的Pattern，我们可以创建一个

PatternStream：
    val input = ...
    val pattern = ...
    
    val patternStream = CEP.pattern(input, pattern)
    val patternStream = CEP.pattern(loginEventStream.keyBy(_.userId), loginFailPattern)

一旦获得PatternStream，我们就可以通过select或flatSelect，从一个Map序列找到我们需要的警告信息。

### 2.1.4 生成Alert ###

#### 2.1.4.1 select ####

select方法需要实现一个PatternSelectFunction，通过select方法来输出需要的警告。它接受一个Map对，包含string/event，其中key为state的名字，event则为真实的Event。

val loginFailDataStream = patternStream
      .select((pattern: Map[String, Iterable[LoginEvent]]) => {
        val first = pattern.getOrElse("begin", null).iterator.next()
        val second = pattern.getOrElse("next", null).iterator.next()
    
        Warning(first.userId, first.eventTime, second.eventTime, "warning")
      })

其返回值仅为1条记录。

#### 2.1.4.2 flatSelect ####

通过实现PatternFlatSelectFunction，实现与select相似的功能。唯一的区别就是flatSelect方法可以返回多条记录，它通过一个`Collector[OUT]`类型的参数来将要输出的数据传递到下游。

[Pattern API]: https://blog.csdn.net/rongyongfeikai2/article/details/83656505