Flink-CEP快速入门

0. 简介 & 使用步骤

简介

• 所谓 CEP，其实就是“复杂事件处理（Complex Event Processing）”的缩写；而 Flink CEP，就是 Flink 实现的一个用于复杂事件处理的库（library）
• 把事件流中的一个个简单事件，通过一定的规则匹配组合起来，这就是“复杂事件”；然后基于这些满足规则的一组组复杂事件进行转换处理，得到想要的结果进行输出

使用步骤

• 复杂事件处理（CEP）的流程可以分成三个步骤：
  1. 定义一个匹配规则
  2. 将匹配规则应用到事件流上，检测满足规则的复杂事件
  3. 对检测到的复杂事件进行处理，得到结果进行输出

// 实体类
public class LoginEvent {
    public String userId;
    public String ipAddress;
    public String eventType;
    public Long timestamp;

    public LoginEvent(String userId, String ipAddress, String eventType, Long timestamp) {
        this.userId = userId;
        this.ipAddress = ipAddress;
        this.eventType = eventType;
        this.timestamp = timestamp;
    }
}


// CEP Demo
public class Demo003 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取登录事件流，并提取时间戳、生成水位线
        SingleOutputStreamOperator<LoginEvent> sourceData = env
                .fromElements(
                        new LoginEvent("user_1", "192.168.0.1", "fail", 2000L),
                        new LoginEvent("user_1", "192.168.0.2", "fail", 3000L),
                        new LoginEvent("user_2", "192.168.1.29", "fail", 4000L),
                        new LoginEvent("user_1", "171.56.23.10", "fail", 5000L),
                        new LoginEvent("user_2", "192.168.1.29", "success", 6000L),
                        new LoginEvent("user_2", "192.168.1.29", "fail", 7000L),
                        new LoginEvent("user_2", "192.168.1.29", "fail", 8000L)

                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<LoginEvent>() {
                                            @Override
                                            public long extractTimestamp(LoginEvent loginEvent, long l) {
                                                return loginEvent.timestamp;
                                            }
                                        }));

        // 1. 定义一个匹配规则：定义 Pattern，连续的三个登录失败事件
        Pattern<LoginEvent, LoginEvent> pattern = Pattern
                .<LoginEvent>begin("first")  // 以第一个登录失败事件开始
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        return "fail".equals(loginEvent.eventType);
                    }
                })
                .next("second")  // 接着是第二个登录失败事件
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        return "fail".equals(loginEvent.eventType);
                    }
                })
                .next("third")  // 接着是第三个登录失败事件
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        return "fail".equals(loginEvent.eventType);
                    }
                });

        // 2. 将 Pattern 应用到流上，检测匹配的复杂事件，得到一个 PatternStream
        PatternStream<LoginEvent> cepPattern = CEP.pattern(sourceData.keyBy(loginEvent -> loginEvent.userId), pattern);

        // 3. 对检测到的复杂事件进行处理：将匹配到的复杂事件选择出来，然后包装成字符串
        SingleOutputStreamOperator<String> select = cepPattern.select(new PatternSelectFunction<LoginEvent, String>() {
            @Override
            public String select(Map<String, List<LoginEvent>> map) throws Exception {
                LoginEvent first = map.get("first").get(0);
                LoginEvent second = map.get("second").get(0);
                LoginEvent third = map.get("third").get(0);
                return first.userId + " 连续三次登录失败！登录时间：" + first.timestamp + ", " + second.timestamp + ", " + third.timestamp;
            }
        });

        select.print();

        env.execute();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

1. 模式API（Pattern API：匹配规则）

单个模式

• 一个模式可以是一个单例或者循环模式。单例模式只接受一个事件，循环模式可以接受多个事件。 在模式匹配表达式中，模式"a b+ c? d"（或者"a"，后面跟着一个或者多个"b"，再往后可选择的跟着一个"c"，最后跟着一个"d"）， a，c?，和 d都是单例模式，b+是一个循环模式

量词

• 单个模式后面可以跟一个“量词”，用来指定循环的次数，单个模式可以包括“单例（singleton）模式”和“循环（looping）模式”，默认是“单例（singleton）模式”，当定义了量词之后，就变成了“循环模式”，可以匹配接收多个事件

循环模式的方法：

• .oneOrMore()
  • 匹配事件出现一次或多次，假设 a 是一个个体模式，a.oneOrMore()表示可以匹配 1 个或多个 a 的事件组合。我们有时会用 a+来简单表示
• .times(times)
  • 匹配事件发生特定次数（times），例如 a.times(3)表示 aaa
• .times(fromTimes，toTimes)
  • 指定匹配事件出现的次数范围，最小次数为fromTimes，最大次数为toTimes。例如a.times(2, 4)可以匹配 aa，aaa 和 aaaa
• .greedy()
  • 只能用在循环模式后，使当前循环模式变得“贪心”（greedy），也就是总是尽可能多地去匹配。例如 a.times(2, 4).greedy()，如果出现了连续 4 个 a，那么会直接把 aaaa 检测出来进行处理，其他任意 2 个 a 是不算匹配事件的
• .optional()
  • 使当前模式成为可选的，也就是说可以满足这个匹配条件，也可以不满足

// 期望出现4次
start.times(4);

// 期望出现0或者4次
start.times(4).optional();

// 期望出现2、3或者4次
start.times(2, 4);

// 期望出现2、3或者4次，并且尽可能的重复次数多
start.times(2, 4).greedy();

// 期望出现0、2、3或者4次
start.times(2, 4).optional();

// 期望出现0、2、3或者4次，并且尽可能的重复次数多
start.times(2, 4).optional().greedy();

// 期望出现1到多次
start.oneOrMore();

// 期望出现1到多次，并且尽可能的重复次数多
start.oneOrMore().greedy();

// 期望出现0到多次
start.oneOrMore().optional();

// 期望出现0到多次，并且尽可能的重复次数多
start.oneOrMore().optional().greedy();

// 期望出现2到多次
start.timesOrMore(2);

// 期望出现2到多次，并且尽可能的重复次数多
start.timesOrMore(2).greedy();

// 期望出现0、2或多次
start.timesOrMore(2).optional();

// 期望出现0、2或多次，并且尽可能的重复次数多
start.timesOrMore(2).optional().greedy();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

条件

限定子类型

• 调用.subtype()方法可以为当前模式增加子类型限制条件

  // 这里 SubEvent 是流中数据类型 Event 的子类型。只有事件是 SubEvent 类型时，才可以满足当前模式 pattern 的匹配条件
  pattern.subtype(SubEvent.class);
  1
  2

简单条件（SimpleCondition）

• 简单条件是最简单的匹配规则，只根据当前事件的特征来决定是否接受它。这在本质上其实就是一个 filter 操作

  start.where(new SimpleCondition<MyEvent>() {
      @Override
      public boolean filter(MyEvent myEvent) throws Exception {
          return ... // 一些判断条件
      }
  })
  1
  2
  3
  4
  5
  6

迭代条件（IterativeCondition）

• 在实际应用中，我们可能需要将当前事件跟之前的事件做对比，才能判断出要不要接受当前事件。这种需要依靠之前事件来做判断的条件，就叫作“迭代条件”（Iterative Condition）

  Pattern.<MyEvent>begin("first")
      .where(new IterativeCondition<MyEvent>() {
          @Override
          public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
              if (!"event1001".equals(myEvent.getEvent())) {
                  return false;
              }
  
              // 根据上下文获取 之前的事件，获取之前满足条件的
              Iterable<MyEvent> myEventIterable = context.getEventsForPattern("first");
              // TODO 处理 之前的事件
  
              return ... // 一些判断条件
          }
      });
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15

组合条件

• 可以多个条件一起使用，当有多个判断逻辑的时候我们可能会用if-else的方式，但组合条件可以在 where()方法后继续接or()方法来组合使用

  Pattern<MyEvent, MyEvent> pattern = Pattern.<MyEvent>begin("first")
                  .where(new IterativeCondition<MyEvent>() {
                      @Override
                      public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
                          return ... // 一些判断条件
                      }
                  }).or(new IterativeCondition<MyEvent>() {
                      @Override
                      public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
                          return ... // 一些判断条件
                      }
                  });
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12

终止条件

• 终止条件的定义是通过调用模式对象的.until()方法来实现的

• ⚠️终止条件只与oneOrMore()或者oneOrMore().optional()结合使用

  Pattern<MyEvent, MyEvent> pattern = Pattern.<MyEvent>begin("first")
                  .where(new IterativeCondition<MyEvent>() {
                      @Override
                      public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
                          return ... // 一些判断条件
                      }
                  }).oneOrMore()
                  .until(new IterativeCondition<MyEvent>() {
                      @Override
                      public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
                          return ... // 一些判断条件
                      }
                  });
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13

模式操作列举

组合模式

连续性

• 将多个个体模式组合起来的完整模式，就叫作“组合模式”

• FlinkCEP支持事件之间如下形式的连续策略：

  1. 严格连续: next()期望所有匹配的事件严格的一个接一个出现，中间没有任何不匹配的事件。
  2. 松散连续: followedBy()忽略匹配的事件之间的不匹配的事件。
  3. 不确定的松散连续: followedByAny()更进一步的松散连续，允许忽略掉一些匹配事件的附加匹配。
  4. notNext():如果不想后面直接连着一个特定事件
  5. notFollowedBy()，如果不想一个特定事件发生在两个事件之间的任何地方

  // 严格连续
  Pattern<Event, ?> strict = start.next("middle").where(...);
  
  // 松散连续
  Pattern<Event, ?> relaxed = start.followedBy("middle").where(...);
  
  // 不确定的松散连续
  Pattern<Event, ?> nonDetermin = start.followedByAny("middle").where(...);
  
  // 严格连续的NOT模式
  Pattern<Event, ?> strictNot = start.notNext("not").where(...);
  
  // 松散连续的NOT模式
  Pattern<Event, ?> relaxedNot = start.notFollowedBy("not").where(...);
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14

• within()方法：指定一个模式应该在一定时间内发生

  // 在十秒钟内，从 event1001 开始到 event1004 结束才算
  Pattern<MyEvent, MyEvent> pattern = Pattern.<MyEvent>begin("first")
                  .where(new IterativeCondition<MyEvent>() {
                      @Override
                      public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
                          return "event1001".equals(myEvent.getEvent());
                      }
                  })
                  .followedBy("second")
                  .where(new IterativeCondition<MyEvent>() {
                      @Override
                      public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
                          return "event1004".equals(myEvent.getEvent());
                      }
                  })
                  .within(Time.seconds(10L));
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

循环模式中的近邻条件

• oneOrMore()、times()等循环模式的默认是松散连续，也就是followedBy()模式

  • .consecutive()：在oneOrMore()、times()等循环模式后面跟上consecutive()表示严格连续（next()）

    // 1. 定义 Pattern，登录失败事件，循环检测 3 次
    Pattern<LoginEvent, LoginEvent> pattern = Pattern
        .<LoginEvent>begin("fails")
        .where(new SimpleCondition<LoginEvent>() {
            @Override
            public boolean filter(LoginEvent loginEvent) throws Exception {
                return loginEvent.eventType.equals("fail");
            }
        }).times(3).consecutive();
    1
    2
    3
    4
    5
    6
    7
    8
    9

  • .allowCombinations():在oneOrMore()、times()等循环模式后面跟上allowCombinations()表示不确定的松散连续（followedByAny()）

模式组

• 也可以定义一个模式序列作为begin，followedBy，followedByAny和next的条件。这个模式序列在逻辑上会被当作匹配的条件， 并且返回一个GroupPattern，可以在GroupPattern上使用oneOrMore()，times(#ofTimes)， times(#fromTimes, #toTimes)，optional()，consecutive()，allowCombinations()。

  Pattern<Event, ?> start = Pattern.begin(
      Pattern.<Event>begin("start").where(...).followedBy("start_middle").where(...)
  );
  
  // 严格连续
  Pattern<Event, ?> strict = start.next(
      Pattern.<Event>begin("next_start").where(...).followedBy("next_middle").where(...)
  ).times(3);
  
  // 松散连续
  Pattern<Event, ?> relaxed = start.followedBy(
      Pattern.<Event>begin("followedby_start").where(...).followedBy("followedby_middle").where(...)
  ).oneOrMore();
  
  // 不确定松散连续
  Pattern<Event, ?> nonDetermin = start.followedByAny(
      Pattern.<Event>begin("followedbyany_start").where(...).followedBy("followedbyany_middle").where(...)
  ).optional();
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18

匹配后跳过策略

对于一个给定的模式，同一个事件可能会分配到多个成功的匹配上。为了控制一个事件会分配到多少个匹配上，你需要指定跳过策略AfterMatchSkipStrategy。 有五种跳过策略，如下：

• NO_SKIP: 不跳过
• SKIP_TO_NEXT: 跳至下一个
• SKIP_PAST_LAST_EVENT: 跳过所有子匹配
• SKIP_TO_FIRST: 跳至第一个
• SKIP_TO_LAST: 跳至最后一个

例如，给定一个模式b+ c和一个数据流b1 b2 b3 c，不同跳过策略之间的不同如下：

skipToNext

// 配置跳过策略：skipToNext模式
AfterMatchSkipStrategy skipStrategy = AfterMatchSkipStrategy.skipToNext();
// 将跳过策略加入到模式中
Pattern<MyEvent, MyEvent> pattern = Pattern.<MyEvent>begin("first", skipStrategy)
    .where(new IterativeCondition<MyEvent>() {
        @Override
        public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
            return ... // 一些判断条件
        }
    });
1
2
3
4
5
6
7
8
9
10

skipToFirst(patternName)

// 配置跳过策略：skipToFirst模式，参数传模式名称
AfterMatchSkipStrategy skipStrategy = AfterMatchSkipStrategy.skipToFirst("first");
Pattern<MyEvent, MyEvent> pattern = Pattern.<MyEvent>begin("first", skipStrategy)
    .where(new IterativeCondition<MyEvent>() {
        @Override
        public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
            return "event1001".equals(myEvent.getEvent());
        }
    }).oneOrMore()
    .followedBy("second")
    .where(new IterativeCondition<MyEvent>() {
        @Override
        public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
            return "event1003".equals(myEvent.getEvent());
        }
    })
    .followedBy("thrid")
    .where(new IterativeCondition<MyEvent>() {
        @Override
        public boolean filter(MyEvent myEvent, Context<MyEvent> context) throws Exception {
            return "event1004".equals(myEvent.getEvent());
        }
    });
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

2. 检测模式（检测满足规则的复杂事件）

将模式应用到流上

• 调用 CEP 类的静态方法.pattern()，将数据流（DataStream）和模式（Pattern）作为两个参数传入
• DataStream，也可以通过 keyBy 进行按键分区得到 KeyedStream，接下来对复杂事件的检测就会针对不同的 key 单独进行了

DataStream<Event> inputStream = ...
Pattern<Event, ?> pattern = ...
PatternStream<Event> patternStream = CEP.pattern(inputStream, pattern);
1
2
3

处理匹配事件

匹配事件的选择提取（select）

PatternSelectFunction

• 处理匹配事件最简单的方式，就是从 PatternStream 中直接把匹配的复杂事件提取出来，包装成想要的信息输出，这个操作就是“选择”（select）

  Pattern.<MyEvent>begin("first").where(...);  
  
  // 处理匹配事件
  cepPattern.select(new PatternSelectFunction<MyEvent, String>() {
      @Override
      public String select(Map<String, List<MyEvent>> map) throws Exception {
          // first 是 Pattern 的 name 字符串
          List<MyEvent> first = map.get("first");
          return ...  // 处理匹配事件逻辑
      }
  });
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11

PatternFlatSelectFunction

• .flatSelect()，传入的参数是一个PatternFlatSelectFunction。这是 PatternSelectFunction 的“扁平化”版本；内部需要实现一个 flatSelect()方法，

• 它与之前 select()的不同就在于没有返回值，而是多了一个收集器（Collector）参数 collector，通过调用 collector.collet()方法就可以实现多次发送输出数据了

  cepPattern.flatSelect(new PatternFlatSelectFunction<MyEvent, String>() {
      @Override
      public void flatSelect(Map<String, List<MyEvent>> map, Collector<String> collector) throws Exception {
          // 处理匹配事件逻辑
      }
  });
  1
  2
  3
  4
  5
  6

匹配事件的通用处理（process）

• 自 1.8 版本之后，Flink CEP 引入了对于匹配事件的通用检测处理方式，那就是直接调用PatternStream 的.process()方法，传入一个 PatternProcessFunction。这看起来就像是我们熟悉的处理函数（process function），它也可以访问一个上下文（Context），进行更多的操作。

• PatternProcessFunction 功能更加丰富、调用更加灵活，可以完全覆盖其他接口，也就成为了目前官方推荐的处理方式。事实上，PatternSelectFunction 和 PatternFlatSelectFunction在 CEP 内部执行时也会被转换成 PatternProcessFunction

• Context context：上下文

• collector.collect()：调用此方法实现发送输出数据

  cepPattern.process(new PatternProcessFunction<MyEvent, String>() {
      @Override
      public void processMatch(Map<String, List<MyEvent>> map, Context context, Collector<String> collector) throws Exception {
          // 处理匹配事件逻辑
      }
  });
  1
  2
  3
  4
  5
  6

处理超时事件

• 在 Flink CEP 中 ， 提 供 了 一 个 专 门 捕 捉 超 时 的 部 分 匹 配 事 件 的 接 口 ， 叫 作TimedOutPartialMatchHandler。这个接口需要实现一个 processTimedOutMatch()方法，可以将超时的、已检测到的部分匹配事件放在一个 Map 中，作为方法的第一个参数；方法的第二个参数则是 PatternProcessFunction 的上下文 Context。所以这个接口必须与 PatternProcessFunction结合使用，对处理结果的输出则需要利用侧输出流来进行

  PatternStream<MyEvent> cepPattern = CEP.pattern(myEventData.keyBy(myEvent -> myEvent.getUserId()), pattern);
  // 测流
  OutputTag<String> outputTag = new OutputTag<String>("time_out"){};
  // 超时数据处理
  SingleOutputStreamOperator<String> processData = cepPattern.process(new MyPatternProcessFunction());
  
  
  
  // 数据处理，处理匹配成功数据，处理超时数据
  public static class MyPatternProcessFunction extends PatternProcessFunction<MyEvent, String>
      implements TimedOutPartialMatchHandler<MyEvent> {
  
      @Override
      public void processMatch(Map<String, List<MyEvent>> map, Context context, Collector<String> collector) throws Exception {
          // 匹配成功逻辑处理
      }
  
      @Override
      public void processTimedOutMatch(Map<String, List<MyEvent>> map, Context context) throws Exception {
          // 超时逻辑处理，将数据写入到测输出流中
          OutputTag<String> outputTag = new OutputTag<String>("time_out"){};
          String str = ...  // 逻辑处理
          context.output(outputTag, str);
      }
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25

Maven

<properties>
    <flink.version>1.13.0flink.version>
    <java.version>1.8java.version>
    <scala.binary.version>2.12scala.binary.version>
    <slf4j.version>1.7.30slf4j.version>
properties>


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


<dependency>
    <groupId>org.apache.flinkgroupId>
    <artifactId>flink-javaartifactId>
    <version>${flink.version}version>
dependency>
<dependency>
    <groupId>org.apache.flinkgroupId>
    <artifactId>flink-streaming-java_${scala.binary.version}artifactId>
    <version>${flink.version}version>
dependency>
<dependency>
    <groupId>org.apache.flinkgroupId>
    <artifactId>flink-clients_${scala.binary.version}artifactId>
    <version>${flink.version}version>
dependency>
<dependency>
    <groupId>org.apache.flinkgroupId>
    <artifactId>flink-connector-kafka_${scala.binary.version}artifactId>
    <version>${flink.version}version>
dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35