大数据之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）。模式的定义主要就是两部分内容：

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

事件发生的顺序即“近邻关系“。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>

定义登录事件

package flinkCEPDemo;

/**
 * @Title: LoginEvent
 * @Author lizhe
 * @Package flinkCEPDemo
 * @Date 2024/6/23 12:46
 * @description:定义登录事件
 */
public class LoginEvent {
    public String userId;
    public String ipAddress;
    public String eventType;
    public Long timestamp;

    public LoginEvent() {
    }

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

    @Override
    public String toString() {
        return "LoginEvent{" +
                "userId='" + userId + '\'' +
                ", ipAddress='" + ipAddress + '\'' +
                ", eventType='" + eventType + '\'' +
                ", timestamp=" + timestamp +
                '}';
    }
}

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

package flinkCEPDemo;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Title: LoginFailDetectExample
 * @Author lizhe
 * @Package flinkCEPDemo
 * @Date 2024/6/23 12:45
 * @description:检测用户行为，如果连续三次登录失败，就输出报警信息。
 */
public class LoginFailDetectExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //获取数据流
        SingleOutputStreamOperator<LoginEvent> loginEventSingleOutputStreamOperator = 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 element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }));
        //定义复杂事件模式
        Pattern<LoginEvent, LoginEvent> loginEventPattern = Pattern.<LoginEvent>begin("first")//第一次登录失败事件
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value, Context<LoginEvent> ctx) throws Exception {
                        return value.eventType.equals("fail");
                    }
                })
                .next("second")//紧跟着第二次登录失败事件
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value, Context<LoginEvent> ctx) throws Exception {
                        return value.eventType.equals("fail");
                    }
                })
                .next("third")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value, Context<LoginEvent> ctx) throws Exception {
                        return value.eventType.equals("fail");
                    }
                });
        //将模式应用到数据流上，检测复杂事件
        PatternStream<LoginEvent> patternStream = CEP.pattern(loginEventSingleOutputStreamOperator.keyBy(event -> event.userId), loginEventPattern);

        //将检测到的事件提取，进行处理得到报警信息输出
        SingleOutputStreamOperator<String> warnStream = patternStream.select(
                new PatternSelectFunction<LoginEvent, String>() {
                    @Override
                    public String select(Map<String, List<LoginEvent>> pattern) throws Exception {
                        //提取复杂事件中的三次登录失败
                        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(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value, Context<LoginEvent> ctx) throws Exception {
                        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(new IterativeCondition<Event>() { @Override
public boolean filter(LoginEvent value, Context<LoginEvent> ctx) throws Exception{
return value.user.startsWith("A");
}
});

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

迭代条件

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

middle.oneOrMore()
.where(new IterativeCondition<Event>() { @Override
public boolean filter(Event value, Context<Event> ctx) throws Exception {
// 事件中的 user 必须以 A 开头
if (!value.user.startsWith("A")) { return false;
}

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(new SimpleCondition<LoginEvent>() { @Override
public boolean filter(LoginEvent loginEvent) throws Exception { 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）操作。

匹配事件的选择提取

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

SingleOutputStreamOperator<String> warnStream = patternStream.select(
                new PatternSelectFunction<LoginEvent, String>() {
                    @Override
                    public String select(Map<String, List<LoginEvent>> pattern) throws Exception {
                        //提取复杂事件中的三次登录失败
                        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(new PatternSelectFunction<LoginEvent, String>() { 
    @Override
public String select(Map<String, List<LoginEvent>> map) throws Exception {
// 只有一个模式，匹配到了 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(new PatternFlatSelectFunction<LoginEvent, String>() { @Override
public void flatSelect(Map<String, List<LoginEvent>> map, Collector<String> out) throws Exception {
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 的功能。

匹配事件的通用处理

直接调用PatternStream 的.process()方法，传入一个 PatternProcessFunction。PatternProcessFunction 功能更加丰富、调用更加灵活，可以完全覆盖其他接口。

代码示例

patternStream.process(new PatternProcessFunction<LoginEvent, String>() { @Override
public void processMatch(Map<String, List<LoginEvent>> map, Context ctx, Collector<String> out) throws Exception {
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 结合使用，对处理结果的输出则需要利用侧输出流来进行。

class MyPatternProcessFunction extends PatternProcessFunction<Event, String>
implements TimedOutPartialMatchHandler<Event> {
// 正常匹配事件的处理
@Override
public void processMatch(Map<String, List<Event>> match, Context ctx, Collector<String> out) throws Exception{
...
}

// 超时部分匹配事件的处理
@Override
public void processTimedOutMatch(Map<String, List<Event>> match, Context ctx) throws Exception{
Event startEvent = match.get("start").get(0);
OutputTag<Event> outputTag = new OutputTag<Event>("time-out"){}; 
ctx.output(outputTag, startEvent);
}
}

2、使用 PatternTimeoutFunction

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

// 定义一个侧输出流标签，用于标识超时侧输出流
OutputTag<String> timeoutTag = new OutputTag<String>("timeout"){};

// 将匹配到的，和超时部分匹配的复杂事件提取出来，然后包装成提示信息输出
SingleOutputStreamOperator<String> resultStream = patternStream.select(timeoutTag,
// 超时部分匹配事件的处理
new PatternTimeoutFunction<Event, String>() { 
    @Override
public String timeout(Map<String, List<Event>> pattern, long timeoutTimestamp) throws Exception {
Event event = pattern.get("start").get(0); return "超时：" + event.toString();
}
},
// 正常匹配事件的处理
new PatternSelectFunction<Event, String>() { 
    @Override
public String select(Map<String, List<Event>> pattern) throws Exception
{
...
}
}
);

// 将正常匹配和超时部分匹配的处理结果流打印输出
resultStream.print("matched");
resultStream.getSideOutput(timeoutTag).print("timeout");

3、应用示例

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

创建订单实体类

package flinkCEPDemo;

/**
 * @Title: OrderEvent
 * @Author lizhe
 * @Package flinkCEPDemo
 * @Date 2024/6/23 19:57
 * @description:定义订单实体
 */
public class OrderEvent {
    public String userId;
    public String orderId;
    public  String eventType;
    public long timestamp;

    public OrderEvent() {
    }

    public OrderEvent(String userId, String orderId, String eventType, long timestamp) {
        this.userId = userId;
        this.orderId = orderId;
        this.eventType = eventType;
        this.timestamp = timestamp;
    }

    @Override
    public String toString() {
        return "OrderEvent{" +
                "userId='" + userId + '\'' +
                ", orderId='" + orderId + '\'' +
                ", eventType='" + eventType + '\'' +
                ", timestamp=" + timestamp +
                '}';
    }
}

代码示例

package flinkCEPDemo;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Title: OrderTimeOutDetectExample
 * @Author lizhe
 * @Package flinkCEPDemo
 * @Date 2024/6/23 19:59
 * @description: 订单超时检测
 */
public class OrderTimeOutDetectExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //获取数据流
        SingleOutputStreamOperator<OrderEvent> orderEventDataStream = env.fromElements(
                new OrderEvent("user_1", "order_1", "create", 1000L),
                new OrderEvent("user_2", "order_2", "create", 2000L),
                new OrderEvent("user_1", "order_1", "modify", 10 * 1000L),
                new OrderEvent("user_1", "order_1", "pay", 60 * 1000L),
                new OrderEvent("user_2", "order_3", "create", 10 * 60 * 1000L),
                new OrderEvent("user_2", "order_3", "pay", 20 * 60 * 1000L)

        ).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ZERO)
        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
            @Override
            public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                return element.timestamp;
            }
        }));
        //定义模式
        Pattern<OrderEvent, OrderEvent> orderEventOrderEventPattern = Pattern.<OrderEvent>begin("create")
                .where(new IterativeCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value, Context<OrderEvent> ctx) throws Exception {
                        return value.eventType.equals("create");
                    }
                })
                .followedBy("pay")
                .where(new IterativeCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value, Context<OrderEvent> ctx) throws Exception {
                        return value.eventType.equals("pay");
                    }
                })
                .within(Time.minutes(15));
        //模式应用到事件流上
        PatternStream<OrderEvent> orderEventPatternStream = CEP.pattern(orderEventDataStream.keyBy(value -> value.orderId), orderEventOrderEventPattern);
        //定义侧输出流
        OutputTag<String> timeoutTag = new OutputTag<String>("timeout") {
        };
        //将完全匹配和部分超时匹配的复杂事件提取出来进行处理
        SingleOutputStreamOperator<String> reslut = orderEventPatternStream.process(new OrderPayMatch());
        reslut.print("payed");
        reslut.getSideOutput(timeoutTag).print("timeout");
        env.execute();
    }
    public static class OrderPayMatch extends PatternProcessFunction<OrderEvent,String> implements TimedOutPartialMatchHandler<OrderEvent>{

        @Override
        public void processMatch(Map<String, List<OrderEvent>> match, Context ctx, Collector<String> out) throws Exception {
            //获取当前的支付事件
            OrderEvent payEvent = match.get("pay").get(0);
            out.collect("用户"+payEvent.userId+"订单"+payEvent.orderId+"已支付");
        }

        @Override
        public void processTimedOutMatch(Map<String, List<OrderEvent>> match, Context ctx) throws Exception {
            OrderEvent createEvent = match.get("create").get(0);
            OutputTag<String> timeoutTag = new OutputTag<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 = new OutputTag<String>("late-data"){};

SingleOutputStreamOperator<ComplexEvent> result = patternStream
.sideOutputLateData(lateDataOutputTag)	// 将迟到数据输出到侧输出流
.select(
// 处理正常匹配数据
new PatternSelectFunction<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 事件，就又满足了新的连续三次登录失败，需要再次报警。