当前位置: 首页 > news >正文

Flink CEP(三)pattern动态更新

        线上运行的CEP中肯定经常遇到规则变更的情况,如果每次变更时都将任务重启、重新发布是非常不优雅的。尤其在营销或者风控这种对实时性要求比较高的场景,如果规则窗口过长(一两个星期),状态过大,就会导致重启时间延长,期间就会造成一些想要处理的异常行为不能及时发现。

1.实现分析

  • 外部加载:通常规则引擎会有专门的规则管理模块,提供用户去创建自己的规则,对于Flink任务来说需要到外部去加载规则
  • 动态更新:需要提供定时去检测规则是否变更
  • 历史状态清理:在模式匹配中是一系列NFAState 的不断变更,如果规则发生变更,需要清理历史状态
  • API:需要对外提供易用的API

2.代码实现

       首先实现一个用户API。

package cep.functions;import java.io.Serializable;import org.apache.flink.api.common.functions.Function;import cep.pattern.Pattern;/*** @author StephenYou* Created on 2023-07-23* Description: 动态Pattern接口(用户调用API)不区分key*/
public interface DynamicPatternFunction<T> extends Function, Serializable {/**** 初始化* @throws Exception*/public void init() throws Exception;/*** 注入新的pattern* @return*/public Pattern<T,T> inject() throws Exception;/*** 一个扫描周期:ms* @return*/public long getPeriod() throws Exception;/*** 规则是否发生变更* @return*/public boolean isChanged() throws Exception;
}

        希望上述API的调用方式如下。

//正常调用CEP.pattern(dataStream,pattern);//动态PatternCEP.injectionPattern(dataStream, new UserDynamicPatternFunction())

        所以需要修改CEP-Lib源码

        b.增加injectionPattern函数。

public class CEP {/**** Dynamic injection pattern function * @param input* @param dynamicPatternFunction* @return* @param <T>*/public static <T> PatternStream<T> injectionPattern throws Exception (DataStream<T> input,DynamicPatternFunction<T> dynamicPatternFunction){return new PatternStream<>(input, dynamicPatternFunction); }
}

        增加PatternStream构造函数,因为需要动态更新,所以有必要传进去整个函数。

public class PatternStream<T> {PatternStream(final DataStream<T> inputStream, DynamicPatternFunction<T> dynamicPatternFunction) throws Exception {this(PatternStreamBuilder.forStreamAndPatternFunction(inputStream, dynamicPatternFunction));}
}

        修改PatternStreamBuilder.build, 增加调用函数的过程。

        final CepOperator<IN, K, OUT> operator = null;if (patternFunction == null ) {operator = new CepOperator<>(inputSerializer,isProcessingTime,nfaFactory,comparator,pattern.getAfterMatchSkipStrategy(),processFunction,lateDataOutputTag);} else {operator = new CepOperator<>(inputSerializer,isProcessingTime,patternFunction,comparator,null,processFunction,lateDataOutputTag);}

        增加对应的CepOperator构造函数。

    public CepOperator(final TypeSerializer<IN> inputSerializer,final boolean isProcessingTime,final DynamicPatternFunction patternFunction,@Nullable final EventComparator<IN> comparator,@Nullable final AfterMatchSkipStrategy afterMatchSkipStrategy,final PatternProcessFunction<IN, OUT> function,@Nullable final OutputTag<IN> lateDataOutputTag) {super(function);this.inputSerializer = Preconditions.checkNotNull(inputSerializer);this.patternFunction = patternFunction;this.isProcessingTime = isProcessingTime;this.comparator = comparator;this.lateDataOutputTag = lateDataOutputTag;if (afterMatchSkipStrategy == null) {this.afterMatchSkipStrategy = AfterMatchSkipStrategy.noSkip();} else {this.afterMatchSkipStrategy = afterMatchSkipStrategy;}this.nfaFactory = null;}

        加载Pattern,构造NFA

    @Overridepublic void open() throws Exception {super.open();timerService =getInternalTimerService("watermark-callbacks", VoidNamespaceSerializer.INSTANCE, this);//初始化if (patternFunction != null) {patternFunction.init();Pattern pattern = patternFunction.inject();afterMatchSkipStrategy = pattern.getAfterMatchSkipStrategy();boolean timeoutHandling = getUserFunction() instanceof TimedOutPartialMatchHandler;nfaFactory = NFACompiler.compileFactory(pattern, timeoutHandling);long period = patternFunction.getPeriod();// 注册定时器检测规则是否变更if (period > 0) {getProcessingTimeService().registerTimer(timerService.currentProcessingTime() + period, this::onProcessingTime);}}nfa = nfaFactory.createNFA();nfa.open(cepRuntimeContext, new Configuration());context = new ContextFunctionImpl();collector = new TimestampedCollector<>(output);cepTimerService = new TimerServiceImpl();// metricsthis.numLateRecordsDropped = metrics.counter(LATE_ELEMENTS_DROPPED_METRIC_NAME);}

        状态清理一共分为两块: 匹配状态数据清理、定时器清理;

        进行状态清理:

    @Overridepublic void processElement(StreamRecord<IN> element) throws Exception {if (patternFunction != null) {// 规则版本更新if (needRefresh.value() < refreshVersion.get()) {//清除状态computationStates.clear();elementQueueState.clear();partialMatches.releaseCacheStatisticsTimer();//清除定时器Iterable<Long> registerTime = registerTimeState.get();if (registerTime != null) {Iterator<Long> iterator = registerTime.iterator();while (iterator.hasNext()) {Long l = iterator.next();//删除定时器timerService.deleteEventTimeTimer(VoidNamespace.INSTANCE, l);timerService.deleteProcessingTimeTimer(VoidNamespace.INSTANCE, l);//状态清理iterator.remove();}}//更新当前的版本needRefresh.update(refreshVersion.get());}}
}

        上面是在处理每条数据时,清除状态和版本。接下来要进行状态和版本的初始化。

    @Overridepublic void initializeState(StateInitializationContext context) throws Exception {super.initializeState(context);//初始化状态if (patternFunction != null) {/*** 两个标识位状态*/refreshFlagState = context.getOperatorStateStore().getUnionListState(new ListStateDescriptor<Integer>("refreshFlagState", Integer.class));if (context.isRestored()) {if (refreshFlagState.get().iterator().hasNext()) {refreshVersion = new AtomicInteger(refreshFlagState.get().iterator().next());}} else {refreshVersion = new AtomicInteger(0);}needRefresh = context.getKeyedStateStore().getState(new ValueStateDescriptor<Integer>("needRefreshState", Integer.class, 0));}
}

3.测试验证

        设置每10s变更一次Pattern。

 PatternStream patternStream = CEP.injectionPattern(source, new TestDynamicPatternFunction());patternStream.select(new PatternSelectFunction<Tuple3<String, Long, String>, Map>() {@Overridepublic Map select(Map map) throws Exception {map.put("processingTime", System.currentTimeMillis());return map;}}).print();env.execute("SyCep");}public static class TestDynamicPatternFunction implements DynamicPatternFunction<Tuple3<String, Long, String>> {public TestDynamicPatternFunction() {this.flag = true;}boolean flag;int time = 0;@Overridepublic void init() throws Exception {flag = true;}@Overridepublic Pattern<Tuple3<String, Long, String>, Tuple3<String, Long, String>> inject()throws Exception {// 2种patternif (flag) {Pattern pattern = Pattern.<Tuple3<String, Long, String>>begin("start").where(new IterativeCondition<Tuple3<String, Long, String>>() {@Overridepublic boolean filter(Tuple3<String, Long, String> value,Context<Tuple3<String, Long, String>> ctx) throws Exception {return value.f2.equals("success");}}).times(1).followedBy("middle").where(new IterativeCondition<Tuple3<String, Long, String>>() {@Overridepublic boolean filter(Tuple3<String, Long, String> value,Context<Tuple3<String, Long, String>> ctx) throws Exception {return value.f2.equals("fail");}}).times(1).next("end");return pattern;} else {Pattern pattern = Pattern.<Tuple3<String, Long, String>>begin("start2").where(new IterativeCondition<Tuple3<String, Long, String>>() {@Overridepublic boolean filter(Tuple3<String, Long, String> value,Context<Tuple3<String, Long, String>> ctx) throws Exception {return value.f2.equals("success2");}}).times(2).next("middle2").where(new IterativeCondition<Tuple3<String, Long, String>>() {@Overridepublic boolean filter(Tuple3<String, Long, String> value,Context<Tuple3<String, Long, String>> ctx) throws Exception {return value.f2.equals("fail2");}}).times(2).next("end2");return pattern;}}@Overridepublic long getPeriod() throws Exception {return 10000;}@Overridepublic boolean isChanged() throws Exception {flag = !flag ;time += getPeriod();System.out.println("change pattern : " + time);return true;}}

打印结果:符合预期

4.源码地址

感觉有用的话,帮忙点个小星星。^_^

 GitHub - StephenYou520/SyCep: CEP 动态Pattern

相关文章:

Flink CEP(三)pattern动态更新

线上运行的CEP中肯定经常遇到规则变更的情况&#xff0c;如果每次变更时都将任务重启、重新发布是非常不优雅的。尤其在营销或者风控这种对实时性要求比较高的场景&#xff0c;如果规则窗口过长&#xff08;一两个星期&#xff09;&#xff0c;状态过大&#xff0c;就会导致重启…...

抽象工厂模式(C++)

定义 提供一个接口,让该接口负责创建一系列“相关或者相互依赖的对象”&#xff0c;无需指定它们具体的类。 使用场景 在软件系统中&#xff0c;经常面临着“一系列相互依赖的对象”的创建工作;同时,由于需求的变化&#xff0c;往往存在更多系列对象的创建工作。如何应对这种…...

程序员面试金典17.*

文章目录 17.01 不用加号的加法17.04 消失的数字17.05字母与数字17.06 2出现的次数17.07 婴儿名字17.08 马戏团人塔17.09 第k个数17.10 主要元素17.11 单词距离17.12 BiNode17.13 恢复空格&#xff08;未做&#xff0c;字典树dp&#xff09;17.14 最小K个数17.15 最长单词17.16…...

【瑞吉外卖项目复写】基本部分复写笔记

Day1 瑞吉外卖项目概述 mysql的数据源配置 spring:datasource:druid:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://localhost:3306/regie?serverTimezoneAsia/Shanghai&useUnicodetrue&characterEncodingutf-8&zeroDateTimeBehaviorconvertTo…...

用html+javascript打造公文一键排版系统15:一键删除所有空格

现在我们来实现一键删除所有空格的功能。 一、使用原有的代码来实现&#xff0c;测试效果并不理想 在这之前我们已经为String对象编写了一个使用正则表达式来删除所有空格的方法&#xff1a; //功能&#xff1a;删除字符串中的所有空格 //记录&#xff1a;20230726创建 Stri…...

苍穹外卖day12(完结撒花)——工作台+Spring_Apche_POI+导出运营数据Excel报表

工作台——需求分析与设计 产品原型 接口设计 工作台——代码导入 将提供的代码导入对应的位置。 工作台——功能测试 Apache POI_介绍 应用场景 Apache POI_入门案例 导入坐标 <!-- poi --><dependency><groupId>org.apache.poi</groupId><ar…...

SQL与NoSQL概念(详细介绍!!)

先搞清楚全称 SQL全称为Structured query language &#xff0c;即结构化查询语言&#xff0c;可以把他理解为一门特殊的编程语言。 那么nosql是什么意思呢&#xff1f;这里的no并不仅是not&#xff0c;而是not only的意思&#xff0c;所以nosql全称应该是Not Only Structure…...

node debian 镜像 new Date 获取时间少 8 小时问题

问题 在 node debian 镜像中&#xff0c;用 (new Date()).getHours() 与系统时间&#xff08;东 8 区&#xff09;少了 8 小时 系统时间 $ node > (new Date()).getHours() 11容器中的时间 $ node > (new Date()).getHours() 3原 Dockerfile FROM node:20.5-bullsey…...

【N32L40X】学习笔记13-软件IIC读写EEPROM AT24C02

AT24C02 8个字节每页,累计32个页 通讯频率MAX 400K AT24C02大小 2K 芯片地址 对于at24c02 A2A1A0 这三个引脚没有使用 写时序 由于设备在写周期中不会产生ACK恢复&#xff0c;因此这可用于确定周期何时完成&#xff08;此特性可用于最大限度地提高总线吞吐量&#xff09;…...

JVM 调优

点击下方关注我&#xff0c;然后右上角点击...“设为星标”&#xff0c;就能第一时间收到更新推送啦~~~ JVM调优是一项重要的任务&#xff0c;可以提高Java应用程序的性能和稳定性。掌握JVM调优需要深入了解JVM的工作原理、参数和配置选项&#xff0c;以及历史JVM参数的调整和优…...

DP-GAN剩余代码

在前面计算完损失后&#xff0c;该进行更新&#xff1a; 1&#xff1a;netEMA是模型的生成器&#xff1a; 遍历生成器的state_dict&#xff0c;将每一个键对应的值乘以EMA_decay。 接着根据当前迭代步数计算num_upd&#xff0c;每1000,2500,10000代倍数就执行一次。 当num…...

在word的文本框内使用Endnote引用文献,如何保证引文编号按照上下文排序

问题 如下图所示&#xff0c;我在word中插入了一个文本框&#xff08;为了插图&#xff09;&#xff0c;然后文本框内有引用&#xff0c;结果endnote自动将文本框内的引用优先排序&#xff0c;变成文献[1]了&#xff0c;而事实上应该是[31]。请问如何能让文本框内的排序也自动…...

SpringBoot项目上传至服务器

1.服务器安装JDK1.8 通过包管理器安装 2.服务器安装数据库 参考链接&#xff1a; CentOS 7 通过 yum 安装 MariaDB - 知乎 1. 安装之后没有密码&#xff0c;所以需要设置密码&#xff0c;使用下面的语句 set password for rootlocalhost password(111111); 2.在数据库中建…...

C++中实现多线程的三种方式

目录 1 背景2 方法 1 背景 力扣1116题 打印零和奇偶数。 2 方法 方法1&#xff1a;原子操作 class ZeroEvenOdd { private:int n;atomic<int> flag 0; public:ZeroEvenOdd(int n) {this->n n;}// printNumber(x) outputs "x", where x is an integer.…...

程序员副业指南:怎样实现年入10w+的目标?

大家好&#xff0c;这里是程序员晚枫&#xff0c;全网同名。 今天给大家分享一个大家都感兴趣的话题&#xff1a;程序员可以做什么副业&#xff0c;年入十万&#xff1f; 01 推荐 程序员可以从事以下副业&#xff0c;以获得一年收入10w&#xff1a; 兼职编程&#xff1a;可…...

excel 计算 分位值

_XLFN.QUARTILE.EXC(Result 1!G:G,2) 和 PERCENTILE 都可以用来计算一组数据的分位数&#xff0c;但是它们的计算方式略有不同。 _XLFN.QUARTILE.EXC(Result 1!G:G,2) 是 Excel 中的一个函数&#xff0c;在计算一个数据集的四分位数时使用。其中&#xff0c;第一个参数 Result…...

mongodb-windows-x86_64-4.4.23-signed.msi

...

一个SpringBoot 项目能处理多少请求?

这篇文章带大家盘一个读者遇到的面试题哈。 根据读者转述&#xff0c;面试官的原问题就是&#xff1a;一个 SpringBoot 项目能同时处理多少请求&#xff1f; 不知道你听到这个问题之后的第一反应是什么。 我大概知道他要问的是哪个方向&#xff0c;但是对于这种只有一句话的…...

Shell编程基础(十)读取多行文本到数组 写入多行文本到文件

读取多行文本到数组 & 写入多行文本到文件 读取多行文本到数组写入多行文本到文件 读取多行文本到数组 创建一个文本文件&#xff0c;内容如下 1 zhangsan 男 10 2 liis 女 12 3 wangwu 男 17读取这个文件中所有人的信息 #!/bin/bash while read u do echo $u done <…...

MyBatis学习笔记2

CRUD 1.namespace namespace中的包名要和mapper接口的包名一致&#xff01; 2.select 选择查询语句 id&#xff1a;就是对应的namespace中的方法名&#xff1b; resultType:Sql语句执行的返回值&#xff01; parameterType&#xff1a;参数类型 增删改必须提交事务&…...

零门槛NAS搭建:WinNAS如何让普通电脑秒变私有云?

一、核心优势&#xff1a;专为Windows用户设计的极简NAS WinNAS由深圳耘想存储科技开发&#xff0c;是一款收费低廉但功能全面的Windows NAS工具&#xff0c;主打“无学习成本部署” 。与其他NAS软件相比&#xff0c;其优势在于&#xff1a; 无需硬件改造&#xff1a;将任意W…...

mongodb源码分析session执行handleRequest命令find过程

mongo/transport/service_state_machine.cpp已经分析startSession创建ASIOSession过程&#xff0c;并且验证connection是否超过限制ASIOSession和connection是循环接受客户端命令&#xff0c;把数据流转换成Message&#xff0c;状态转变流程是&#xff1a;State::Created 》 St…...

2025 后端自学UNIAPP【项目实战:旅游项目】6、我的收藏页面

代码框架视图 1、先添加一个获取收藏景点的列表请求 【在文件my_api.js文件中添加】 // 引入公共的请求封装 import http from ./my_http.js// 登录接口&#xff08;适配服务端返回 Token&#xff09; export const login async (code, avatar) > {const res await http…...

GitHub 趋势日报 (2025年06月08日)

&#x1f4ca; 由 TrendForge 系统生成 | &#x1f310; https://trendforge.devlive.org/ &#x1f310; 本日报中的项目描述已自动翻译为中文 &#x1f4c8; 今日获星趋势图 今日获星趋势图 884 cognee 566 dify 414 HumanSystemOptimization 414 omni-tools 321 note-gen …...

微信小程序云开发平台MySQL的连接方式

注&#xff1a;微信小程序云开发平台指的是腾讯云开发 先给结论&#xff1a;微信小程序云开发平台的MySQL&#xff0c;无法通过获取数据库连接信息的方式进行连接&#xff0c;连接只能通过云开发的SDK连接&#xff0c;具体要参考官方文档&#xff1a; 为什么&#xff1f; 因为…...

Android15默认授权浮窗权限

我们经常有那种需求&#xff0c;客户需要定制的apk集成在ROM中&#xff0c;并且默认授予其【显示在其他应用的上层】权限&#xff0c;也就是我们常说的浮窗权限&#xff0c;那么我们就可以通过以下方法在wms、ams等系统服务的systemReady()方法中调用即可实现预置应用默认授权浮…...

OPENCV形态学基础之二腐蚀

一.腐蚀的原理 (图1) 数学表达式&#xff1a;dst(x,y) erode(src(x,y)) min(x,y)src(xx,yy) 腐蚀也是图像形态学的基本功能之一&#xff0c;腐蚀跟膨胀属于反向操作&#xff0c;膨胀是把图像图像变大&#xff0c;而腐蚀就是把图像变小。腐蚀后的图像变小变暗淡。 腐蚀…...

HDFS分布式存储 zookeeper

hadoop介绍 狭义上hadoop是指apache的一款开源软件 用java语言实现开源框架&#xff0c;允许使用简单的变成模型跨计算机对大型集群进行分布式处理&#xff08;1.海量的数据存储 2.海量数据的计算&#xff09;Hadoop核心组件 hdfs&#xff08;分布式文件存储系统&#xff09;&a…...

FFmpeg:Windows系统小白安装及其使用

一、安装 1.访问官网 Download FFmpeg 2.点击版本目录 3.选择版本点击安装 注意这里选择的是【release buids】&#xff0c;注意左上角标题 例如我安装在目录 F:\FFmpeg 4.解压 5.添加环境变量 把你解压后的bin目录&#xff08;即exe所在文件夹&#xff09;加入系统变量…...

Bean 作用域有哪些?如何答出技术深度?

导语&#xff1a; Spring 面试绕不开 Bean 的作用域问题&#xff0c;这是面试官考察候选人对 Spring 框架理解深度的常见方式。本文将围绕“Spring 中的 Bean 作用域”展开&#xff0c;结合典型面试题及实战场景&#xff0c;帮你厘清重点&#xff0c;打破模板式回答&#xff0c…...