大数据之Flink（六）

17、Flink CEP

17.1、概念

17.1.1、CEP

CEP是“复杂事件处理（Complex Event Processing）”的缩写；而 Flink CEP，就是 Flink 实现的一个用于复杂事件处理的库（library）。

总结起来，复杂事件处理（CEP）的流程可以分成三个步骤：

（1） 定义一个匹配规则

（2） 将匹配规则应用到事件流上，检测满足规则的复杂事件

（3） 对检测到的复杂事件进行处理，得到结果进行输出

输入是不同形状的事件流，我们可以定义一个匹配规则：在圆形后面紧跟着三角形。那么将这个规则应用到输入流上，就可以检测到三组匹配的复杂事件。它们构成了一个新的“复杂事件流”，流中的数据就变成了一组一组的复杂事件，每个数据都包含了一个圆形和一个三角形。接下来，我们就可以针对检测到的复杂事件，处理之后输出一个提示或报警信息了。

CEP 是针对流处理而言的，分析的是低延迟、频繁产生的事件流。它的主要目的， 就是在无界流中检测出特定的数据组合，掌握数据中重要的高阶特征。

17.1.2、模式（Pattern）

CEP 的第一步所定义的匹配规则，我们可以把它叫作“模式”（Pattern）。模式的定义主要就是两部分内容：

1. 每个简单事件的特征
2. 简单事件之间的组合关系

事件发生的顺序即“近邻关系“。CEP 做的事其实就是在流上进行模式匹配。根据模式的近邻关系条件不同，可以检测连续的事件或不连续但先后发生的事件；模式还可能有时间的限制，如果在设定时间范围内没有满足匹配条件，就会导致模式匹配超时。

17.1.3、应用场景

⚫ 风险控制

设定一些行为模式，可以对用户的异常行为进行实时检测。当一个用户行为符合了异常行为模式，比如短时间内频繁登录并失败、大量下单却不支付（刷单），就可以向用户发送通知信息，或是进行报警提示、由人工进一步判定用户是否有违规操作的嫌疑。这样就可以有效地控制用户个人和平台的风险。

⚫ 用户画像

利用 CEP 可以用预先定义好的规则，对用户的行为轨迹进行实时跟踪，从而检测出具有特定行为习惯的一些用户，做出相应的用户画像。基于用户画像可以进行精准营销，即对行为匹配预定义规则的用户实时发送相应的营销推广；这与目前很多企业所做的精准推荐原理是一样的。

⚫ 运维监控

对于企业服务的运维管理，可以利用 CEP 灵活配置多指标、多依赖来实现更复杂的监控模式。

17.2、简易代码

17.2.1、引入依赖

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

定义登录事件

packageflinkCEPDemo;/**
 * @Title: LoginEvent
 * @Author lizhe
 * @Package flinkCEPDemo
 * @Date 2024/6/23 12:46
 * @description:定义登录事件
 */publicclassLoginEvent{publicString userId;publicString ipAddress;publicString eventType;publicLong timestamp;publicLoginEvent(){}publicLoginEvent(String userId,String ipAddress,String eventType,Long timestamp){this.userId = userId;this.ipAddress = ipAddress;this.eventType = eventType;this.timestamp = timestamp;}@OverridepublicStringtoString(){return"LoginEvent{"+"userId='"+ userId +'\''+", ipAddress='"+ ipAddress +'\''+", eventType='"+ eventType +'\''+", timestamp="+ timestamp +'}';}}

检测用户行为，如果连续三次登录失败，就输出报警信息。

packageflinkCEPDemo;importorg.apache.flink.api.common.eventtime.SerializableTimestampAssigner;importorg.apache.flink.api.common.eventtime.WatermarkStrategy;importorg.apache.flink.cep.CEP;importorg.apache.flink.cep.PatternSelectFunction;importorg.apache.flink.cep.PatternStream;importorg.apache.flink.cep.pattern.Pattern;importorg.apache.flink.cep.pattern.conditions.IterativeCondition;importorg.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importjava.time.Duration;importjava.util.List;importjava.util.Map;/**
 * @Title: LoginFailDetectExample
 * @Author lizhe
 * @Package flinkCEPDemo
 * @Date 2024/6/23 12:45
 * @description:检测用户行为，如果连续三次登录失败，就输出报警信息。
 */publicclassLoginFailDetectExample{publicstaticvoidmain(String[] args)throwsException{StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);//获取数据流SingleOutputStreamOperator<LoginEvent> loginEventSingleOutputStreamOperator = env.fromElements(newLoginEvent("user_1","192.168.0.1","fail",2000L),newLoginEvent("user_1","192.168.0.2","fail",3000L),newLoginEvent("user_2","192.168.1.29","fail",4000L),newLoginEvent("user_1","171.56.23.10","fail",5000L),newLoginEvent("user_2","192.168.1.29","success",6000L),newLoginEvent("user_2","192.168.1.29","fail",7000L),newLoginEvent("user_2","192.168.1.29","fail",8000L)).assignTimestampsAndWatermarks(WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(newSerializableTimestampAssigner<LoginEvent>(){@OverridepubliclongextractTimestamp(LoginEvent element,long recordTimestamp){return element.timestamp;}}));//定义复杂事件模式Pattern<LoginEvent,LoginEvent> loginEventPattern =Pattern.<LoginEvent>begin("first")//第一次登录失败事件.where(newIterativeCondition<LoginEvent>(){@Overridepublicbooleanfilter(LoginEvent value,Context<LoginEvent> ctx)throwsException{return value.eventType.equals("fail");}}).next("second")//紧跟着第二次登录失败事件.where(newIterativeCondition<LoginEvent>(){@Overridepublicbooleanfilter(LoginEvent value,Context<LoginEvent> ctx)throwsException{return value.eventType.equals("fail");}}).next("third").where(newIterativeCondition<LoginEvent>(){@Overridepublicbooleanfilter(LoginEvent value,Context<LoginEvent> ctx)throwsException{return value.eventType.equals("fail");}});//将模式应用到数据流上，检测复杂事件PatternStream<LoginEvent> patternStream =CEP.pattern(loginEventSingleOutputStreamOperator.keyBy(event -> event.userId), loginEventPattern);//将检测到的事件提取，进行处理得到报警信息输出SingleOutputStreamOperator<String> warnStream = patternStream.select(newPatternSelectFunction<LoginEvent,String>(){@OverridepublicStringselect(Map<String,List<LoginEvent>> pattern)throwsException{//提取复杂事件中的三次登录失败LoginEvent firstLoginEvent = pattern.get("first").get(0);LoginEvent secondLoginEvent = pattern.get("second").get(0);LoginEvent thirdLoginEvent = pattern.get("third").get(0);return firstLoginEvent.userId +"连续三次登录失败！登录时间分别为"+ firstLoginEvent.timestamp +","+ secondLoginEvent.timestamp +","+ thirdLoginEvent.timestamp;}});
        warnStream.print();
        env.execute();}}

17.3、模式API（Pattern API）

17.3.1、个体模式

一个匹配规则就可以表达成先后发生的一个个简单事件，按顺序串联组合在一起。这里的每一个简单事件并不是任意选取的，也需要有一定的条件规则；所以我们就把每个简单事件的匹配规则，叫作“个体模式”（Individual Pattern）。

每一个登录失败事件的选取规则，就都是一个个体模式

.next("second")//紧跟着第二次登录失败事件.where(newIterativeCondition<LoginEvent>(){@Overridepublicbooleanfilter(LoginEvent value,Context<LoginEvent> ctx)throwsException{return value.eventType.equals("fail");}})

体模式需要一个“过滤条件”，用来指定具体的匹配规则。这个条件一般是通过调用.where()方法来实现的。

个体模式增加一个“量词”（quantifier），就能够让它进行循环匹配，接收多个事件。

17.3.1.1、量词

个体模式后面可以跟一个“量词”，用来指定循环的次数。在循环模式中，对同样特征的事件可以匹配多次。比如我们定义个体模式为“匹配形状为三角形的事件”，再让它循环多次，就变成了“匹配连续多个三角形的事件”。注意这里的“连续”，只要保证前后顺序即可，中间可以有其他事件，所以是“宽松近邻”关系。

• .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 次
pattern.times(4);// 匹配事件出现 4 次，或者不出现
pattern.times(4).optional();// 匹配事件出现 2, 3 或者 4 次
pattern.times(2,4);// 匹配事件出现 2, 3 或者 4 次，并且尽可能多地匹配
pattern.times(2,4).greedy();// 匹配事件出现 2, 3, 4 次，或者不出现
 pattern.times(2,4).optional().greedy();//匹配事件出现 1 次或多次
pattern.oneOrMore();// 匹配事件出现 1 次或多次，并且尽可能多地匹配
pattern.oneOrMore().greedy();// 匹配事件出现 1 次或多次，或者不出现
pattern.oneOrMore().optional();// 匹配事件出现 1 次或多次，或者不出现；并且尽可能多地匹配
pattern.oneOrMore().optional().greedy();// 匹配事件出现 2 次或多次
pattern.timesOrMore(2);// 匹配事件出现 2 次或多次，并且尽可能多地匹配
pattern.timesOrMore(2).greedy();// 匹配事件出现 2 次或多次，或者不出现
pattern.timesOrMore(2).optional()// 匹配事件出现 2 次或多次，或者不出现；并且尽可能多地匹配
pattern.timesOrMore(2).optional().greedy();

17.3.1.2、条件

对于条件的定义，主要是通过调用 Pattern 对象的.where()方法来实现的，主要可以分为简单条件、迭代条件、组合条件、终止条件几种类型。

• 简单条件

匹配事件的user 属性以“A”开头

pattern.where(newIterativeCondition<Event>(){@Overridepublicbooleanfilter(LoginEvent value,Context<LoginEvent> ctx)throwsException{return value.user.startsWith("A");}});

返回为true为符合匹配规则，false为不符合匹配规则。

• 迭代条件

上下文 Context。调用这个上下文的.getEventsForPattern()方法，传入一个模式名称，就可以拿到这个模式中之前已匹配到的所有数据了。

middle.oneOrMore().where(newIterativeCondition<Event>(){@Overridepublicbooleanfilter(Event value,Context<Event> ctx)throwsException{// 事件中的 user 必须以 A 开头if(!value.user.startsWith("A")){returnfalse;}int sum = value.amount;// 获取当前模式之前已经匹配的事件，求所有事件 amount 之和for(Event event : ctx.getEventsForPattern("middle")){ sum += event.amount;}// 在总数量小于 100 时，当前事件满足匹配规则，可以匹配成功return sum <100;}});

迭代条件能够获取已经匹配的事件，如果自身又是循环模式（比如量词oneOrMore），那么两者结合就可以捕获自身之前接收的数据，据此来判断是否接受当前事件。这个功能非常强大，我们可以由此实现更加复杂的需求，比如可以要求“只有大于之前数据的平均值，才接受当前事件”。

另外迭代条件中的上下文Context 也可以获取到时间相关的信息，比如事件的时间戳和当前的处理时（processing time）。

• 组合条件.where()后面再接一个.where()相当于就是多个条件的“逻辑与”（AND）.where()后加一个.or()相当于多个条件的逻辑或（OR）
• 终止条件指定一个“终止条件”（Stop Condition），表示遇到某个特定事件时当前模式就不再继续循环匹配了。终止条件的定义是通过调用模式对象的.until() 方法来实现的， 同样传入一个IterativeCondition 作为参数。需要注意的是，终止条件只与 oneOrMore() 或者oneOrMore().optional()结合使用。因为在这种循环模式下，我们不知道后面还有没有事件可以匹配，只好把之前匹配的事件作为状态缓存起来继续等待，这等待无穷无尽；如果一直等下去， 缓存的状态越来越多，最终会耗尽内存。所以这种循环模式必须有个终点，当.until()指定的条件满足时，循环终止，这样就可以清空状态释放内存了。

17.3.2、组合模式

按一定的顺序把个体模式组合起来——模式序列。

Pattern<Event,?> pattern =Pattern.<Event>begin("start").where(...).next("next").where(...).followedBy("follow").where(...)...

1、初始模式

Pattern 的静态方法.begin()来创建,注意begin前要有泛型。如下所示：

Pattern<Event,?> start =Pattern.<Event>begin("start");

2、近邻条件

• 严格近邻匹配的事件严格地按顺序一个接一个出现，中间不会有任何其他事件。.next()
• 宽松近邻两个匹配的事件之间可以有其他不匹配的事件出现。.followedBy()
• 非确定性宽松近邻所谓“非确定性”是指可以重复使用之前已经匹配过的事件；这种近邻条件下匹配到的不同复杂事件，可以以同一个事件作为开始，即图中圆圈可以跟所有的三角进行匹配。 3、其他限制条件
• .notNext()

表示前一个模式匹配到的事件后面，不能紧跟着某种事件。

• .notFollowedBy()

表示前一个模式匹配到的事件后面， 不会出现某种事件。这里需要注意， 由于notFollowedBy()是没有严格限定的；流数据不停地到来，我们永远不能保证之后“不会出现某种事件”。所以一个模式序列不能以 notFollowedBy()结尾，这个限定条件主要用来表示“两个事件中间不会出现某种事件”。

• .within()传入一个时间参数，这是模式序列中第一个事件到最后一个事件之间的最大时间间隔，只有在这期间成功匹配的复杂事件才是有效的。一个模式序列中只能有一个时间限制，调用.within()的位置不限；如果多次调用则会以最小的那个时间间隔为准

// 严格近邻条件Pattern<Event,?> strict = start.next("middle").where(...);// 宽松近邻条件Pattern<Event,?> relaxed = start.followedBy("middle").where(...);// 非确定性宽松近邻条件Pattern<Event,?> nonDetermin = start.followedByAny("middle").where(...);// 不能严格近邻条件Pattern<Event,?> strictNot = start.notNext("not").where(...);// 不能宽松近邻条件Pattern<Event,?> relaxedNot = start.notFollowedBy("not").where(...);// 时间限制条件
middle.within(Time.seconds(10));

对于定义了量词（如 oneOrMore()、times()）的循环模式，默认内部采用的是宽松近邻,即当循环匹配多个事件时，它们中间是可以有其他不匹配事件的；相当于用单例模式分别定义、再用 followedBy()连接起来。.consecutive()为循环模式中的匹配事件增加严格的近邻条件，保证所有匹配事件是严格连续的。优化17.2的代码

//  定义 Pattern，登录失败事件，循环检测 3 次Pattern<LoginEvent,LoginEvent> pattern =Pattern.<LoginEvent>begin("fails").where(newSimpleCondition<LoginEvent>(){@Overridepublicbooleanfilter(LoginEvent loginEvent)throwsException{return loginEvent.eventType.equals("fail");}}).times(3).consecutive();

除严格近邻外，也可以为循环模式中的事件指定非确定性宽松近邻条件，表示可以重复使用 已 经 匹 配 的 事 件 。 这 需 要 调 用 .allowCombinations() 方 法 来 实 现 ， 实 现 的 效 果与.followedByAny()相同。

17.3.3、模式组

非常复杂的场景中，可能需要划分多个“阶段”，每个“阶段”又有一连串的匹配规则。为了应对这样的需求，Flink CEP 允许我们以“嵌套”的方式来定义模式。

我们用 begin()、next()、followedBy()、followedByAny()这样的“连接词”来组合个体模式，这些方法的参数就是一个个体模式的名称；而现在它们可以直接以一个模式序列作为参数，就将模式序列又一次连接组合起来了。这样得到的就是一个“模式组”（Groups of Patterns）

// 以模式序列作为初始模式Pattern<Event,?> start =Pattern.begin(Pattern.<Event>begin("start_start").where(...).followedBy("start_middle").where(...));// 在 start 后定义严格近邻的模式序列，并重复匹配两次Pattern<Event,?> strict = start.next(Pattern.<Event>begin("next_start").where(...).followedBy("next_middle").where(...)).times(2);// 在 start 后定义宽松近邻的模式序列，并重复匹配一次或多次Pattern<Event,?> relaxed = start.followedBy(Pattern.<Event>begin("followedby_start").where(...).followedBy("followedby_middle").where(...)).oneOrMore();//在 start 后定义非确定性宽松近邻的模式序列，可以匹配一次，也可以不匹配Pattern<Event,?> nonDeterminRelaxed = start.followedByAny(Pattern.<Event>begin("followedbyany_start").where(...).followedBy("followedbyany_middle").where(...)).optional();

17.3.4、匹配后跳过策略

如果对循环模式增加了.greedy()的限制，那么就会“尽可能多地”匹配事件，这样就可以砍掉那些子集上的匹配了。不过这种方式还是略显简单粗暴，如果我们想要精确控制事件的匹配应该跳过哪些情况，那就需要制定另外的策略了。

Pattern.begin("start",AfterMatchSkipStrategy.noSkip()).where(...)...

匹配后跳过策略 AfterMatchSkipStrategy 是一个抽象类，它有多个具体的实现，可以通过调用对应的静态方法来返回对应的策略实例。

我们如果输入事件序列“a a a b”——这里为了区分前后不同的 a 事件，可以记作“a1 a2 a3 b”——那么应该检测到 6 个匹配结果：（a1 a2 a3 b），（a1 a2 b），（a1 b），（a2 a3 b），（a2 b），（a3 b）。如果在初始模式的量词.oneOrMore()后加上.greedy()定义为贪心匹配，那么结果就是：

（a1 a2 a3 b），（a2 a3 b），（a3 b），每个事件作为开头只会出现一次。接下来我们讨论不同跳过策略对匹配结果的影响：

• 不跳过（NO_SKIP）

代码调用AfterMatchSkipStrategy.noSkip()。这是默认策略，所有可能的匹配都会输出。所以这里会输出完整的 6 个匹配。

• 跳至下一个（SKIP_TO_NEXT）

代码调用 AfterMatchSkipStrategy.skipToNext()。找到一个 a1 开始的最大匹配之后，跳过a1 开始的所有其他匹配，直接从下一个 a2 开始匹配起。当然 a2 也是如此跳过其他匹配。最终得到（a1 a2 a3 b），（a2 a3 b），（a3 b）。可以看到，这种跳过策略跟使用.greedy()效果是相同的。

• 跳过所有子匹配（SKIP_PAST_LAST_EVENT）

代码调用AfterMatchSkipStrategy.skipPastLastEvent()。找到 a1 开始的匹配（a1 a2 a3 b）之后，直接跳过所有 a1 直到 a3 开头的匹配，相当于把这些子匹配都跳过了。最终得到（a1 a2 a3 b），这是最为精简的跳过策略。

• 跳至第一个（SKIP_TO_FIRST[a]）

代码调用AfterMatchSkipStrategy.skipToFirst(“a”)，这里传入一个参数，指明跳至哪个模式的第一个匹配事件。找到 a1 开始的匹配（a1 a2 a3 b）后，跳到以最开始一个 a（也就是 a1） 为开始的匹配，相当于只留下 a1 开始的匹配。最终得到（a1 a2 a3 b），（a1 a2 b），（a1 b）。

• 跳至最后一个（SKIP_TO_LAST[a]）

代码调用AfterMatchSkipStrategy.skipToLast(“a”)，同样传入一个参数，指明跳至哪个模式的最后一个匹配事件。找到 a1 开始的匹配（a1 a2 a3 b）后，跳过所有 a1、a2 开始的匹配，跳到以最后一个 a（也就是 a3）为开始的匹配。最终得到（a1 a2 a3 b），（a3 b）。

17.4、模式的检测处理

17.4.1、模式应用到流上

利用 Pattern API 定义好模式还只是整个复杂事件处理的第一步，接下来还需要将模式应用到事件流上、检测提取匹配的复杂事件并定义处理转换的方法，最终得到想要的输出信息。

PatternStream<LoginEvent> patternStream =CEP.pattern(loginEventSingleOutputStreamOperator.keyBy(event -> event.userId), loginEventPattern);

模式中定义的复杂事件，发生是有先后顺序的，这里“先后”的判断标准取决于具体的时间语义。

// 可选的事件比较器EventComparator<Event> comparator =...PatternStream<Event> patternStream =CEP.pattern(input, pattern, comparator);

17.4.2、处理匹配事件

PatternStream 的转换操作主要可以分成两种：简单便捷的选择提取（select）操作，和更加通用、更加强大的处理（process）操作。

1. 匹配事件的选择提取处理匹配事件最简单的方式，就是从 PatternStream 中直接把匹配的复杂事件提取出来， 包装成想要的信息输出，这个操作就是“选择”（select）。SingleOutputStreamOperator<String> warnStream = patternStream.select(newPatternSelectFunction<LoginEvent,String>(){@OverridepublicStringselect(Map<String,List<LoginEvent>> pattern)throwsException{//提取复杂事件中的三次登录失败LoginEvent firstLoginEvent = pattern.get("first").get(0);LoginEvent secondLoginEvent = pattern.get("second").get(0);LoginEvent thirdLoginEvent = pattern.get("third").get(0);return firstLoginEvent.userId +"连续三次登录失败！登录时间分别为"+ firstLoginEvent.timestamp +","+ secondLoginEvent.timestamp +","+ thirdLoginEvent.timestamp;}});PatternSelectFunction 里需要实现一个 select()方法，这个方法每当检测到一组匹配的复杂事件时都会调用一次。它以保存了匹配复杂事件的 Map 作为输入，经自定义转换后得到输出信息返回。这里我们假设之前定义的模式序列中，有名为“start”和“middle”的两个个体模式， 于是可以通过这个名称从 Map 中选择提取出对应的事件。注意调用 Map 的.get(key)方法后得到的是一个事件的List；如果个体模式是单例的，那么List 中只有一个元素，直接调用.get(0) 就可以把它取出。对连续登录失败检测的改进patternStream.select(newPatternSelectFunction<LoginEvent,String>(){@OverridepublicStringselect(Map<String,List<LoginEvent>> map)throwsException{// 只有一个模式，匹配到了 3 个事件，放在 List 中LoginEvent first = map.get("fails").get(0);LoginEvent second = map.get("fails").get(1);LoginEvent third = map.get("fails").get(2);return first.userId +" 连续三次登录失败！登录时间："+ first.timestamp+", "+ second.timestamp +", "+ third.timestamp;}}).print("warning");PatternStream 还有一个类似的方法是.flatSelect() ， 传入的参数是一个PatternFlatSelectFunction。与之前 select()的不同就在于没有返回值，而是多了一个收集器（Collector）参数out，通过调用 out.collet()方法就可以实现多次发送输出数据了。代码示例patternStream.flatSelect(newPatternFlatSelectFunction<LoginEvent,String>(){@OverridepublicvoidflatSelect(Map<String,List<LoginEvent>> map,Collector<String> out)throwsException{LoginEvent first = map.get("fails").get(0);LoginEvent second = map.get("fails").get(1);LoginEvent third = map.get("fails").get(2);out.collect(first.userId +" 连续三次登录失败！登录时间："+ first.timestamp +", "+ second.timestamp +", "+ third.timestamp);}}).print("warning");PatternFlatSelectFunction 使用更加灵活，完全能够覆盖 PatternSelectFunction 的功能。
2. 匹配事件的通用处理直接调用PatternStream 的.process()方法，传入一个 PatternProcessFunction。PatternProcessFunction 功能更加丰富、调用更加灵活，可以完全覆盖其他接口。代码示例patternStream.process(newPatternProcessFunction<LoginEvent,String>(){@OverridepublicvoidprocessMatch(Map<String,List<LoginEvent>> map,Context ctx,Collector<String> out)throwsException{LoginEvent first = map.get("fails").get(0);LoginEvent second = map.get("fails").get(1);LoginEvent third = map.get("fails").get(2);out.collect(first.userId +" 连续三次登录失败！登录时间："+ first.timestamp +", "+ second.timestamp +", "+ third.timestamp);}}).print("warning");

17.4.3、处理超时事件

当一个模式上通过

within

加上窗口长度后，部分匹配的事件序列就可能因为超过窗口长度而被丢弃。可以使用

TimedOutPartialMatchHandler

接口 来处理超时的部分匹配。这个接口可以和其它的混合使用。所以往往不应该直接丢弃，而是要输出一个提示或报警信息。这就要求我们有能力捕获并处理超时事件。

1、使用 PatternProcessFunction 的侧输出流

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

classMyPatternProcessFunctionextendsPatternProcessFunction<Event,String>implementsTimedOutPartialMatchHandler<Event>{// 正常匹配事件的处理@OverridepublicvoidprocessMatch(Map<String,List<Event>> match,Context ctx,Collector<String> out)throwsException{...}// 超时部分匹配事件的处理@OverridepublicvoidprocessTimedOutMatch(Map<String,List<Event>> match,Context ctx)throwsException{Event startEvent = match.get("start").get(0);OutputTag<Event> outputTag =newOutputTag<Event>("time-out"){}; 
ctx.output(outputTag, startEvent);}}

2、使用 PatternTimeoutFunction

简化版的PatternSelectFunction ， 它无法直接处理超时事件， 不过我们可以通过调用 PatternStream 的.select()方法时多传入一个 PatternTimeoutFunction 参数来实现这一点。

// 定义一个侧输出流标签，用于标识超时侧输出流OutputTag<String> timeoutTag =newOutputTag<String>("timeout"){};// 将匹配到的，和超时部分匹配的复杂事件提取出来，然后包装成提示信息输出SingleOutputStreamOperator<String> resultStream = patternStream.select(timeoutTag,// 超时部分匹配事件的处理newPatternTimeoutFunction<Event,String>(){@OverridepublicStringtimeout(Map<String,List<Event>> pattern,long timeoutTimestamp)throwsException{Event event = pattern.get("start").get(0);return"超时："+ event.toString();}},// 正常匹配事件的处理newPatternSelectFunction<Event,String>(){@OverridepublicStringselect(Map<String,List<Event>> pattern)throwsException{...}});// 将正常匹配和超时部分匹配的处理结果流打印输出
resultStream.print("matched");
resultStream.getSideOutput(timeoutTag).print("timeout");

3、应用示例

在电商平台中，最终创造收入和利润的是用户下单购买的环节。用户下单的行为可以表明用户对商品的需求，但在现实中，并不是每次下单都会被用户立刻支付。当拖延一段时间后， 用户支付的意愿会降低。所以为了让用户更有紧迫感从而提高支付转化率，同时也为了防范订单支付环节的安全风险，电商网站往往会对订单状态进行监控，设置一个失效时间（比如 15 分钟），如果下单后一段时间仍未支付，订单就会被取消。

创建订单实体类

packageflinkCEPDemo;/**
 * @Title: OrderEvent
 * @Author lizhe
 * @Package flinkCEPDemo
 * @Date 2024/6/23 19:57
 * @description:定义订单实体
 */publicclassOrderEvent{publicString userId;publicString orderId;publicString eventType;publiclong timestamp;publicOrderEvent(){}publicOrderEvent(String userId,String orderId,String eventType,long timestamp){this.userId = userId;this.orderId = orderId;this.eventType = eventType;this.timestamp = timestamp;}@OverridepublicStringtoString(){return"OrderEvent{"+"userId='"+ userId +'\''+", orderId='"+ orderId +'\''+", eventType='"+ eventType +'\''+", timestamp="+ timestamp +'}';}}

代码示例

packageflinkCEPDemo;importorg.apache.flink.api.common.eventtime.SerializableTimestampAssigner;importorg.apache.flink.api.common.eventtime.WatermarkStrategy;importorg.apache.flink.cep.CEP;importorg.apache.flink.cep.PatternStream;importorg.apache.flink.cep.functions.PatternProcessFunction;importorg.apache.flink.cep.functions.TimedOutPartialMatchHandler;importorg.apache.flink.cep.pattern.Pattern;importorg.apache.flink.cep.pattern.conditions.IterativeCondition;importorg.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.streaming.api.functions.ProcessFunction;importorg.apache.flink.streaming.api.windowing.time.Time;importorg.apache.flink.util.Collector;importorg.apache.flink.util.OutputTag;importjava.time.Duration;importjava.util.List;importjava.util.Map;/**
 * @Title: OrderTimeOutDetectExample
 * @Author lizhe
 * @Package flinkCEPDemo
 * @Date 2024/6/23 19:59
 * @description: 订单超时检测
 */publicclassOrderTimeOutDetectExample{publicstaticvoidmain(String[] args)throwsException{StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);//获取数据流SingleOutputStreamOperator<OrderEvent> orderEventDataStream = env.fromElements(newOrderEvent("user_1","order_1","create",1000L),newOrderEvent("user_2","order_2","create",2000L),newOrderEvent("user_1","order_1","modify",10*1000L),newOrderEvent("user_1","order_1","pay",60*1000L),newOrderEvent("user_2","order_3","create",10*60*1000L),newOrderEvent("user_2","order_3","pay",20*60*1000L)).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(newSerializableTimestampAssigner<OrderEvent>(){@OverridepubliclongextractTimestamp(OrderEvent element,long recordTimestamp){return element.timestamp;}}));//定义模式Pattern<OrderEvent,OrderEvent> orderEventOrderEventPattern =Pattern.<OrderEvent>begin("create").where(newIterativeCondition<OrderEvent>(){@Overridepublicbooleanfilter(OrderEvent value,Context<OrderEvent> ctx)throwsException{return value.eventType.equals("create");}}).followedBy("pay").where(newIterativeCondition<OrderEvent>(){@Overridepublicbooleanfilter(OrderEvent value,Context<OrderEvent> ctx)throwsException{return value.eventType.equals("pay");}}).within(Time.minutes(15));//模式应用到事件流上PatternStream<OrderEvent> orderEventPatternStream =CEP.pattern(orderEventDataStream.keyBy(value -> value.orderId), orderEventOrderEventPattern);//定义侧输出流OutputTag<String> timeoutTag =newOutputTag<String>("timeout"){};//将完全匹配和部分超时匹配的复杂事件提取出来进行处理SingleOutputStreamOperator<String> reslut = orderEventPatternStream.process(newOrderPayMatch());
        reslut.print("payed");
        reslut.getSideOutput(timeoutTag).print("timeout");
        env.execute();}publicstaticclassOrderPayMatchextendsPatternProcessFunction<OrderEvent,String>implementsTimedOutPartialMatchHandler<OrderEvent>{@OverridepublicvoidprocessMatch(Map<String,List<OrderEvent>> match,Context ctx,Collector<String> out)throwsException{//获取当前的支付事件OrderEvent payEvent = match.get("pay").get(0);
            out.collect("用户"+payEvent.userId+"订单"+payEvent.orderId+"已支付");}@OverridepublicvoidprocessTimedOutMatch(Map<String,List<OrderEvent>> match,Context ctx)throwsException{OrderEvent createEvent = match.get("create").get(0);OutputTag<String> timeoutTag =newOutputTag<String>("timeout"){};
            ctx.output(timeoutTag,"用户"+createEvent.userId+"订单"+createEvent.orderId +"超时未支付");}}}

17.4.4、处理迟到数据

在 Flink CEP 中沿用了通过设置水位线（watermark）延迟来处理乱序数据的做法。当一个事件到来时，并不会立即做检测匹配处理，而是先放入一个缓冲区（buffer）。缓冲区内的数据，会按照时间戳由小到大排序；当一个水位线到来时，就会将缓冲区中所有时间戳小于水位线的事件依次取出，进行检测匹配。这样就保证了匹配事件的顺序和事件时间的进展一致，处理的顺序就一定是正确的。这里水位线的延迟时间，也就是事件在缓冲区等待的最大时间。

Flink CEP 同样提供了将迟到事件输出到侧输出流的方式： 我们可以基于 PatternStream 直接调用.sideOutputLateData()方法，传入一个 OutputTag，将迟到数据放入侧输出流另行处理。代码中调用方式如下：

PatternStream<Event> patternStream =CEP.pattern(input, pattern);// 定义一个侧输出流的标签OutputTag<String> lateDataOutputTag =newOutputTag<String>("late-data"){};SingleOutputStreamOperator<ComplexEvent> result = patternStream
.sideOutputLateData(lateDataOutputTag)// 将迟到数据输出到侧输出流.select(// 处理正常匹配数据newPatternSelectFunction<Event,ComplexEvent>(){...});// 从结果中提取侧输出流DataStream<String> lateData = result.getSideOutput(lateDataOutputTag);

17.5、CEP 的状态机实现

Flink CEP 的底层工作原理其实与正则表达式是一致的，是一个“非确定有限状态自动机”（Nondeterministic Finite Automaton，NFA）。

如检测用户连续三次登录失败的复杂事件。用 Flink CEP 中的 Pattern API 可以很方便地把它定义出来；如果我们现在不用 CEP，而是用 DataStream API 和处理函数来实现，应该怎么做呢？

这需要设置状态，并根据输入的事件不断更新状态。当然因为这个需求不是很复杂，我们也可以用嵌套的 if-else 条件判断将它实现，不过这样做的代码可读性和扩展性都会很差。更好的方式，就是实现一个状态机。

从初始状态（INITIAL）出发，遇到一个类型为fail 的登录失败事件，就开始进入部分匹配的状态；目前只有一个 fail 事件，我们把当前状态记作 S1。基于 S1 状态，如果继续遇到 fail 事件，那么就有两个 fail 事件，记作 S2。基于 S2 状态如果再次遇到 fail 事件，那么就找到了一组匹配的复杂事件，把当前状态记作 Matched， 就可以输出报警信息了。需要注意的是，报警完毕，需要立即重置状态回 S2；因为如果接下来再遇到 fail 事件，就又满足了新的连续三次登录失败，需要再次报警。