《Flink学习笔记》——第十二章 Flink CEP

12.1 基本概念

12.1.1 CEP是什么

1.什么是CEP？

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

2.那到底什么是“复杂事件处理”呢？

答：就是可以在事件流里，检测到特定的事件组合并进行处理，比如说“连续登录失败”，或者“订单支付超时”等等。具体的处理过程是，把事件流中的一个个简单事件，通过一定的规则匹配组合起来，这就是“复杂事件”；然后基于这些满足规则的一组组复杂事件进行转换处理，得到想要的结果进行输出。

3.CEP的目的是什么？

答：就是在无界流中检测出特定的数据组合，让我们有机会掌握数据中重要的高阶特征

CEP的流程可以分成三个步骤：

（1）定义一个匹配规则

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

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

示例：

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

image-20230408182527254

12.1.2 模式（Pattern）

CEP定义的匹配规则，我们把它叫做模式。

模式的定义主要有两部分：

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

当然，我们也可以进一步扩展模式的功能。比如，匹配检测的时间限制；每个简单事件是否可以重复出现；对于事件可重复出现的模式，遇到一个匹配后是否跳过后面的匹配；等等。

所谓“事件之间的组合关系”，一般就是定义“谁后面接着是谁”，也就是事件发生的顺序。我们把它叫作“近邻关系”。可以定义严格的近邻关系，也就是两个事件之前不能有任何其他事件；也可以定义宽松的近邻关系，即只要前后顺序正确即可，中间可以有其他事件。另外， 还可以反向定义，也就是“谁后面不能跟着谁”。

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

12.1.3 应用场景

CEP主要用于实时流数据的分析处理

  • 风险控制

    设定一些行为模式，可以对用户的异常行为实时检测

  • 用户画像

    精准营销，如客户买了什么那大概率还会买什么，和精准推荐相似

  • 运维监控

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

12.2 快速上手

12.2.1 引入依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-cep</artifactId><version>${flink.version}</version>
</dependency>

12.2.2 一个简单实例

需求：检测用户行为，如果连续三次登录失败，就输出报警信息。很显然，这是一个复杂事件的检测处理，我们可以使用 Flink CEP 来实现

定义一个登录事件POJO类

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class LoginEvent {private String userId;private String ipAddress;private String eventType;private Long timestamp;
}

主函数

public class LoginFailDetect {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);KeyedStream<LoginEvent, String> keyedStream = 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>forMonotonousTimestamps().withTimestampAssigner((SerializableTimestampAssigner<LoginEvent>) (loginEvent, l) -> loginEvent.getTimestamp())).keyBy(LoginEvent::getUserId);// 1. 定义一个模式,连续三次登录失败Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("first").where(new SimpleCondition<LoginEvent>() {@Overridepublic boolean filter(LoginEvent loginEvent) {return "fail".equals(loginEvent.getEventType());}}).next("second").where(new SimpleCondition<LoginEvent>() {@Overridepublic boolean filter(LoginEvent loginEvent) {return "fail".equals(loginEvent.getEventType());}}).next("third").where(new SimpleCondition<LoginEvent>() {@Overridepublic boolean filter(LoginEvent loginEvent) {return "fail".equals(loginEvent.getEventType());}});// 2. 将 Pattern 应用到流上，检测匹配的复杂事件，得到一个 PatternStreamPatternStream<LoginEvent> patternStream = CEP.pattern(keyedStream, pattern);// 3. 将匹配到的复杂事件选择出来，然后包装成字符串报警信息输出patternStream.select((PatternSelectFunction<LoginEvent, String>) map -> {LoginEvent first = map.get("first").get(0);LoginEvent second = map.get("second").get(0);LoginEvent third = map.get("third").get(0);return first.getUserId() + " 连续三次登录失败！登录时间：" +first.getTimestamp() + ", " + second.getTimestamp() + ", " + third.getTimestamp();}).print();env.execute();}
}

输出结果：

user_1 连续三次登录失败！登录时间：2000, 3000, 5000

12.3 模式API

Flink CEP 的核心是复杂事件的模式匹配。Flink CEP 库中提供了 Pattern 类，基于它可以调用一系列方法来定义匹配模式，这就是所谓的模式 API（Pattern API）。Pattern API 可以让我们定义各种复杂的事件组合规则，用于从事件流中提取复杂事件

12.3.1 个体模式

模式就是由一组简单的事件的匹配规则组成，单个事件的匹配规则叫做个体模式。如上面的每一个登录失败事件都是个体模式。

1、基本形式

一般由一个连接词begin、next开始，然后where定义事件特征/匹配规则，并且个体模式通过量词和条件也能接收多个事件。

.begin
.where
.next
.where
2、量词

个体模式后面可以跟一个“量词”，用来指定循环的次数。从这个角度分类，个体模式可以包括“单例（singleton）模式”和“循环（looping）模式”。默认情况下，个体模式是单例模式，匹配接收一个事件；当定义了量词之后，就变成了循环模式，可以匹配接收多个事件。

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

在 Flink CEP 中，可以使用不同的方法指定循环模式，主要有:

  • .oneOrMore（）
  • .times（times）
  • .times（fromTimes，toTimes）
  • .greedy()——贪心模式，尽可能多个匹配
  • .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();// 匹配事件出现 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();
3、条件

对于每个个体模式，匹配事件的核心在于定义匹配条件，也就是选取事件的规则

有以下几种条件类型：

  • 限定子类型：当流中的数据类型为事件类型的子类型时满足条件
  • 简单条件：只根据当前事件的特征来决定是否满足条件
  • 迭代条件：简单条件只能依靠当前事件做判断，能够处理的逻辑有限。迭代条件可以依靠之前的事件做判断。
  • 组合条件：可以通过where后面接一个where或者where后面接or等等，实现多个条件的组合。（其实在一个where里面也可以使用多个if-else来判断，但是代码臃肿可读性差）
  • 终止条件：终止循环模式。注意：一般只与 oneOrMore() 或者 oneOrMore().optional()结合使用。因为只有这两个是可以匹配无穷尽的，所以要有终止条件。

12.3.2 组合模式

就是多个个体模式组成

1、初始模式

所有的组合模式都以begin开始

Pattern.begin()
2、近邻条件

在初始模式之后，我们就可以按照复杂事件的顺序追加模式，组合成模式序列了

Flink CEP 中提供了三种近邻关系:

  • 严格近邻（next）——两个事件紧挨着

  • 宽松近邻（followedBy）——先后顺序正确就行，无序紧挨着出现

  • 非确定性宽松近邻（followedByAny）——可以重复使用之前已经匹配过的事件

    image-20230409014105141

3、其它条件

（1）notNext()

（2）notFollowedBy()

（3）within()

​ 这是模式序列中第一个事件到最后一个事件之间的最大时间间隔，只有在这期间成功匹配的复杂事件才是有效的

4、循环模式中近邻关系

循环模式——个体模式加了量词

在循环模式中，近邻关系同样有三种：严格近邻、宽松近邻以及非确定性宽松近邻

对于定义了量词（如 oneOrMore()、times()）的循环模式，默认内部采用的是宽松近邻，那么可以通过以下方法可以更改近邻关系

（1）consecutive()

​ 如果要为循环模式中的匹配事件增加严格的近邻条件，保证所有匹配事件是严格连续的

（2）allowCombinations()

​ 除严格近邻外，也可以为循环模式中的事件指定非确定性宽松近邻条件，表示可以重复使用已经匹配的事件。

12.3.3 模式组

多个模式的组合、嵌套，返回的类型为GroupPattern，为Pattern的子类型

12.3.4 匹配后跳过策略

如果我们想要精确控制事件的匹配应该跳过哪些情况，就需要制定另外的策略

使用:

// begin的第二个参数传入，默认跳过处理
Pattern.begin("start", AfterMatchSkipStrategy.noSkip())

不同的跳过策略：

（1）不跳过（默认）

AfterMatchSkipStrategy.noSkip()

（2）跳至下一个

AfterMatchSkipStrategy.skipToNext()

（3）跳至所有子匹配

AfterMatchSkipStrategy.skipPastLastEvent()

（4）跳至第一个

AfterMatchSkipStrategy.skipToFirst(“a”)

（5）跳至最后一个

AfterMatchSkipStrategy.skipToLast(“a”)

public class PatternTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);KeyedStream<LoginEvent, String> keyedStream = env.fromElements(new LoginEvent("user1", "192.168.0.1", "a", 2000L),new LoginEvent("user1", "192.168.0.2", "a", 3000L),new LoginEvent("user1", "192.168.1.29", "a", 4000L),new LoginEvent("user1", "171.56.23.10", "b", 5000L)).assignTimestampsAndWatermarks(WatermarkStrategy.<LoginEvent>forMonotonousTimestamps().withTimestampAssigner((SerializableTimestampAssigner<LoginEvent>) (loginEvent, l) -> loginEvent.getTimestamp())).keyBy(LoginEvent::getUserId);// 1. 定义一个模式,连续三次登录失败Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("first", AfterMatchSkipStrategy.noSkip()).where(new SimpleCondition<LoginEvent>() {@Overridepublic boolean filter(LoginEvent loginEvent) {return "a".equals(loginEvent.getEventType());}}).oneOrMore().followedBy("second").where(new SimpleCondition<LoginEvent>() {@Overridepublic boolean filter(LoginEvent loginEvent) {return "b".equals(loginEvent.getEventType());}});// 2. 将 Pattern 应用到流上，检测匹配的复杂事件，得到一个 PatternStreamPatternStream<LoginEvent> patternStream = CEP.pattern(keyedStream, pattern);// 3. 将匹配到的复杂事件选择出来，然后包装成字符串报警信息输出patternStream.select(new PatternSelectFunction<LoginEvent, String>() {@Overridepublic String select(Map<String, List<LoginEvent>> map) throws Exception {return String.format("--------%n first: %s%n second: %s%n--------%n", map.get("first"), map.get("second"));}}).print();env.execute();}
}

12.4 模式的检测处理

将模式应用到事件流上、检测提取匹配的复杂事件并定义处理转换的方法，最终得到想要的输出信息

12.4.1 将模式应用到流上

PatternStream<Event> patternStream = CEP.pattern(DataStream/KeyedStream, Pattern);

​ 模式中定义的复杂事件发生是有先后顺序的，这取决于使用哪种时间语义。对于时间戳相同（事件时间）或是同时到达（处理时间）的事件，我们还可以通过比较器，来进行更精确的排序

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

12.4.2 处理匹配事件

1、匹配事件的选择提取

（1）PatternSelectFunction

DataStream<String> result = patternStream.select(new PatternSelectFunction());

（2）PatternFlatSelectFunction

将匹配到的元素“扁平化”，通过收集器输出

DataStream<String> result = patternStream.select(new PatternFlatSelectFunction());
2、匹配事件的通用处理
patternStream.process(new PatternProcessFunction())

PatternProcessFunction 功能更加丰富、调用更加灵活，可以完全覆盖其他接口，也就成为了目前官方推荐的处理方式。

PatternProcessFunction 中必须实现一个 processMatch()方法；这个方法与之前的 flatSelect()类似，只是多了一个上下文 Context 参数。利用这个上下文可以获取当前的时间信息，比如事件的时间戳（timestamp）或者处理时间（processing time）；还可以调用.output()方法将数据输出到侧输出流。

12.4.3 处理超时事件

比如我们用.within()指定了模式检测的时间间隔，超出这个时间当前这组检测就应该失败了.

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

1、使用 PatternProcessFunction 的侧输出流
class MyPatternProcessFunction extends PatternProcessFunction<Event, String>
implements TimedOutPartialMatchHandler<Event>
2、使用 PatternTimeoutFunction

上文提到的PatternProcessFunction 通过实现TimedOutPartialMatchHandler 接口扩展出了处理超时事件的能力，这是官方推荐的做法

3、应用实例
代码略

12.4.4 处理迟到的数据

patternStream.sideOutputLateData(lateDataOutputTag)	// 将迟到数据输出到侧输出流.select(// 处理正常匹配数据new PatternSelectFunction<Event, ComplexEvent>() {...});

12.5 CEP的状态机实现

image-20230409120754206

举例：

代码略

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/91949.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

智汇云舟亮相中国安防工程商集成商大会

智汇云舟亮相中国安防工程商集成商大会

智汇云舟亮相中国安防工程商集成商大会&#xff0c;以视频孪生驱动安防行业数字化转型 近日&#xff0c;由中国安全防范产品行业协会指导&#xff0c;永泰传媒主办的中国安防工程商&#xff08;系统集成商&#xff09;大会暨第69届中国安防新产品、新技术成果展示在石家庄圆满…
阅读更多...
软件架构设计(一) 软件架构的概念

软件架构设计(一) 软件架构的概念

在讲到软件架构的概念时，首先我们要了解到，架构是在做什么样的事情，它在整个软件开发周期中所属什么样的位置。 之前学习软件工程时，我们学到了开发模型，里面涉及到需求分析，概要设计，详细设计，编码，测试。但事实上，没有提到架构这个东西。 为什么这么重要的东西没…
阅读更多...
场景赋能 数创齐鲁|“山东工行杯”山东省第五届数据应用创新创业大赛正式启幕！

场景赋能 数创齐鲁|“山东工行杯”山东省第五届数据应用创新创业大赛正式启幕！

为深入贯彻落实《中共中央 国务院关于构建数据基础制度 更好发挥数据要素作用的意见》&#xff0c;发挥我省海量数据规模和丰富应用场景优势&#xff0c;推动数字产业化、产业数字化、数据价值化和治理服务数字化&#xff0c;进一步提升公共数据资源开发利用水平&#xff0c;促…
阅读更多...
aac音频格式怎么转mp3？这种转换方法很简单

aac音频格式怎么转mp3？这种转换方法很简单

AAC格式和MP3格式都是数字音频格式&#xff0c;但它们使用不同的音频压缩算法。虽然AAC格式在音质和文件大小方面表现出色&#xff0c;但是不是所有的设备都支持该格式。相比之下&#xff0c;MP3格式的兼容性更好&#xff0c;可以在各种设备和操作系统上播放&#xff0c;因此转…
阅读更多...
基于JAVA SpringBoot互联网就医门诊挂号管理系统

基于JAVA SpringBoot互联网就医门诊挂号管理系统

摘要 随着时代的发展,无线互联网技术的应用和普及给人们的生活带来了极大的改变,现在信息技术不仅可以提高我们的工作效率,还能有效的规避一些错误风险,节约人力成本。我国国民一方面对健康的要求越来越重视了&#xff0c;另一方面现代人的健康问题日益严重&#xff0c;所以医院…
阅读更多...
Blender界面学习03 原点、鼠标所在位置的缩放与旋转

Blender界面学习03 原点、鼠标所在位置的缩放与旋转

物体的坐标原点可以移动 放大缩小时默认是屏幕中央&#xff0c;修改为鼠标在哪儿就缩放哪儿 默认旋转时围绕屏幕的中心 可以修改为指定对象旋转
阅读更多...
echarts饼图点击区块事件

echarts饼图点击区块事件

效果图&#xff1a; 代码&#xff1a; let option {color: pieColors,series: [{name: Access From,type: pie,radius: [36%, 56%],avoidLabelOverlap: false,label: {formatter: params > {// console.log(params)return {color${params.dataIndex}|${params.name}(${par…
阅读更多...
学习高级数据结构：探索平衡树与图的高级算法

学习高级数据结构：探索平衡树与图的高级算法

文章目录 1. 平衡树&#xff1a;维护数据的平衡与高效性1.1 AVL 树&#xff1a;严格的平衡1.2 红黑树&#xff1a;近似平衡 2. 图的高级算法&#xff1a;建模复杂关系与优化2.1 最小生成树&#xff1a;寻找最优连接方式2.2 拓扑排序&#xff1a;解决依赖关系 拓展思考 &#x1…
阅读更多...
java入坑之网络编程

java入坑之网络编程

一、 网络基础知识 1.1网卡 1.2IP地址 1.3端口 1.4保留IP 1.5网络协议 二、UDP 编程 2.1相关概念 计算机通讯&#xff1a;数据从一个IP的port出发&#xff08;发送方&#xff09;&#xff0c;运输到另外一个IP的port&#xff08;接收方&#xff09; UDP&#xff1a;无连接无…
阅读更多...
大模型 Dalle2 学习三部曲（一）clip学习

大模型 Dalle2 学习三部曲（一）clip学习

clip论文比较长48页&#xff0c;但是clip模型本身又比较简单&#xff0c;效果又奇好&#xff0c;正所谓大道至简&#xff0c;我们来学习一下clip论文中的一些技巧&#xff0c;可以让我们快速加深对clip模型的理解&#xff0c;以及大模型对推荐带来革命性的变化。 clip结构 首选…
阅读更多...
时序分解 | MATLAB实现基于SVD奇异值分解的信号分解分量可视化

时序分解 | MATLAB实现基于SVD奇异值分解的信号分解分量可视化

时序分解 | MATLAB实现基于SVD奇异值分解的信号分解分量可视化 目录 时序分解 | MATLAB实现基于SVD奇异值分解的信号分解分量可视化效果一览基本介绍程序设计参考资料 效果一览 基本介绍 SVD分解重构算法&#xff0c;MATLAB程序&#xff0c;奇异值分解 (Singular Value Decompo…
阅读更多...
GPT-3在化学中进行低数据发现是否足够？

GPT-3在化学中进行低数据发现是否足够？

今天介绍一份洛桑联邦理工学院进行的工作&#xff0c;这份工作被发表在化学期刊预印本网站上。 对于这份工作&#xff0c;有兴趣的朋友可以通过我们的国内ChatGPT镜像站进行测试使用&#xff0c;我们的站点并没有针对特定任务进行建设&#xff0c;是通用性质的。 化学领域进行…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023