Flink CEP

1 复杂事件

1.1 引入

在Flink RichFunction&state这篇博文中我们一起学习了下如何结合使用keyBy state和TreeSet在一条无界流中进行全局的分组求top n操作，可以解决一些实时看板相关的业务问题。在Flink BroadcastStream这篇博文中我们也学习到了如何使用广播流来处理监控规则经常变更的程序日志监控业务。

那么现在我们又遇到了一个新的业务需求：判断一个用户在点击了商品之后是否立即进行了下单付款操作，如果是的话将用户名和点击时间以及下单付款时间都输出到控制台。
我们思考了一下，发现继续使用keyBy state操作好像也能够实现这个业务需求，只是我们的判断逻辑和代码都会比较复杂一些。逻辑思路如下

1.将数据整理成UserAction(ip: String, timestamp: Long, name: String, action: String)
2.通过name对UserAction进行keyBy操作，得到KeyedStream[UserAction, String]
3.KeyedStream.process(xxx)此时在每个name分组中，需要一个keyBy state同时保存该name对应的"click"操作的UserAction的信息
  使用一个ValueState[UserAction]来保存"click"信息，"buy"信息不需要保存
4.这个时候碰到的问题就是在每个name分组如何判断"click","buy"的UserAction是有顺序来到该分组
5.针对于4的问题，可以通过ValueState存储的信息来进行判断
  a."click","buy"先后来到该name分组
    "click"来到该name分组时，直接将"click"添加更新进ValueState
    "buy"来到该name分组时，判断ValueState是否有存储"click"
        有则将"click"取出来和当前"buy"整合并发送到下游，且要清空ValueState信息
        没有直接清空ValueState信息，不发送任何数据到下游
  b.如果有非"click","buy"信息来到name分组，均直接清空ValueState信息，不发送任何数据到下游

当我们遇到的业务并不是很复杂的时候使用上边的思路完全可以解决，但是如果要是碰到更复杂一些的业务需求可能判断逻辑就会越来越复杂："click"和"buy"之间可以有"show"操作；"click"和"buy"之间可以有零或者多个"click"操作。

那怎么能够使用更简洁的逻辑和代码来实现这种越来越复杂的问题呢？

1.2 复杂事件

我们先来给这样的业务来定这样一个描述：检测和发现无界事件流中多个记录的关联规则，也就是从无界事件流中得到满足规则的复杂事件。

CEP(Complex Event Processing)就是在无界事件流中检测事件模式，让我们掌握数据中重要的部分。flink CEP是在flink中实现的复杂事件处理库。

01_show.png

目标：从简单事件流中发现一些高阶特征
输入：一个或者多个简单事件构成的事件流
处理：检测简单事件之间的联系，多个事件组合一起符合匹配规则，将该多个事件构成复杂事件
输出：符合规则的复杂事件

2 Pattern API

我们先来认识一下CEP中Pattern模式吧，也就是规则的制定。

val start: Pattern[X, X] = Pattern.begin[X]("start")

Pattern根据模式的组合种类，分为了三种

2.1 个体模式(Individual Patterns)

组成复杂规则的每一个单独的模式定义，就是“个体模式”

start.where(condition: F => Boolean)

量词

个体模式根据接收同一种事件的数量又可以分为“单例模式”和“循环模式“，我们通过一个“量词”来指定接受同一种事件的数量。

start.times(2)                      // 必须2次
start.times(2, 5)                   // 2，3，4或者5次都可以
start.times(2, 5).greedy            // 2，3，4或者5次，并且尽可能的重复匹配
start.times(2).optional             // 0或者2次
start.oneOrMore                     // 1次或者多次
start.timesOrMore(2).optional.greedy// 0, 2或者多次，并且尽可能的重复匹配

条件

个体模式的条件，可以在一个个体模式上判断使用多个条件，只有当条件都满足的情况下才算匹配成功

// 组合条件 (参考sql中的 where or)
.where()    // 条件相连为and
.or()       // 条件相连为or

// 终止条件
.until()    // 当使用了oneOrMore或者oneOrMore.optional时需要进行终止，以便清楚状态

// 迭代条件
.where(condition: (F, Context[F]) => Boolean) // 调用上下文对前边接收的事件进行处理
// ctx.getEventsForPattern("start")

2.2 组合模式(Combining Patterns)

多个个体模式组合起来就形成了一个组合模式

Pattern.begin[X]("start").where(condition: F => Boolean)
        .next("middle").where(condition: F => Boolean)

注意：组合模式必须由一个“开始个体模式”开始

val start: Pattern[X, X] = Pattern.begin[X]("start")

2.3 模式组(Groups of Patterns)

将一个组合模式作为条件嵌套在个体模式里，成为一组模式

3 近邻

当我们在对事件流进行复杂事件处理，有时我们需要两个事件在流中必须相连，有时只要两个事件出现在流中不必须相连中间可以间隔一个其他事件。也就是我们对多个事件组成规则严格性的宽容度，那么如何来表达这个容忍度呢？使用近邻。

严格近邻：所有事件严格按照顺序进行，中间没有任何不匹配的事件。使用next()指定

对于模式"a next b"，事件流[a, c, b, d]，没有匹配

02_yange.png

宽松近邻：允许中间出现不匹配事件。使用followedBy()指定

对于模式"a followedBy b"，事件流[a, c, b, d]，匹配为 [a, b]

03_kuansong.png

非确定宽松近邻：一个匹配的事件能够再次使用。使用followedByAny()指定

对于模式"a followedByAny b"，事件流[a, c, b1, b2]，匹配为 [a, b1],[a, b2]

04_feiqueding_kuansong.png

4 示例

此处我们使用flink CEP来实现 1.1引入 中提到的业务需求

1 CEP并不包含在flink中，使用前需要自己导入cep jar包

    <!-- 使用cep需要引入该jar包，
        scala语言编写要导入 artifactId前缀为 flink-cep-scala -->
    <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-cep -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-cep-scala_2.12</artifactId>
        <version>1.9.1</version>
    </dependency>

2 定义Pattern模式且必须要以Pattern.begin("xxx")开始，"xxx"为别名，在后边PatternStream.select()中获取符合模式的数据时会使用到。

val pattern = Pattern.begin[T]("start").where(...).next("middle").where(...)

    // 定义一个cep pattern模式。此处复杂的事务为：用户click后马上进行buy操作
    val pattern: Pattern[UserAction, UserAction] =
      Pattern.begin[UserAction]("start").where(new SimpleCondition[UserAction] {
        override def filter(t: UserAction): Boolean = t.action.equals("click")
      })
        .next("middle").where(new SimpleCondition[UserAction] {
        override def filter(t: UserAction): Boolean = t.action.equals("buy")
      })

    // 获取一个普通的流
    val input: KeyedStream[UserAction, String] = env.addSource(
        new FlinkKafkaConsumer010[String]("test", new SimpleStringSchema(), initProps()))
      .map { line => 
        val strs: Array[String] = line.split(",")
        UserAction(strs(0), strs(1).toLong, strs(2), strs(3)) // 将记录转换为UserAction类型
      }.keyBy(_.name)

    // 将我们定义好的cep pattern应用于这个普通的流
    val patternStream: PatternStream[UserAction] = CEP.pattern(input, pattern)

    // 通过select算子获取符合pattern的事务数据，并打印结果
    patternStream.select(new PatternSelectFunction[UserAction, String] {
      override def select(map: util.Map[String, util.List[UserAction]]): String = {
        val click: UserAction = map.get("start").iterator().next()
        val buy: UserAction = map.get("middle").iterator().next()
        // 打印用户的名称，点击和购买的时间
        s"name: ${click.name}, click: ${click.timestamp}, buy: ${buy.timestamp}"
      }
    }).print()

05_cep_demo.png

示例代码中在Pattern指定时，我们使用next()来表示需要得到是符合严格近邻的复杂事件，所以"click","order","buy"这样的复杂事件并不符合模式也不会被输出打印到控制台。

此博文，我们只是简单的介绍了下CEP及其使用方式，但是CEP的知识远不远不止这些。后边有机会的话我们再一起深入学习，此处就不进行展开了。
下一篇博文我们就来一起学习一下flink中的三种时间语义和窗口机制吧。