二、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)
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。
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 序列找到我们需要的警告信息。
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 条记录。
flatSelect
通过实现 PatternFlatSelectFunction，实现与 select 相似的功能。唯一的区别就是 flatSelect 方法可以返回多条记录，它通过一个 Collector[OUT]类型的参数来将要输出的数据传递到下游。
超时事件的处理
通过 within 方法，我们的 parttern 规则将匹配的事件限定在一定的窗口范围内。当有超过窗口时间之后到达的 event，我们可以通过在 select 或 flatSelect 中，实现PatternTimeoutFunction 和 PatternFlatTimeoutFunction 来处理这种情况。
val patternStream: PatternStream[Event] = CEP.pattern(input, pattern) val outputTag = OutputTag[String]("side-output") val result: SingleOutputStreamOperator[ComplexEvent] = patternStream.select (outputTag){ (pattern: Map[String, Iterable[Event]], timestamp: Long) => TimeoutEvent () } { pattern: Map[String, Iterable[Event]] => ComplexEvent() } val timeoutResult: DataStream<TimeoutEvent> = result.getSideOutput(outputTag)
