flink 加载外部cep规则

2023-10-19 14:32
文章标签 加载 规则 flink 外部 cep

本文主要是介绍flink 加载外部cep规则，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

主程序代码：

package cepengine.App;import cepengine.domain.Event;
import cepengine.domain.InputEventSchema;
import cepengine.domain.OutputEventSchema;
import groovy.lang.GroovyClassLoader;
import groovy.lang.GroovyObject;
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.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;import java.io.File;
import java.util.List;
import java.util.Map;
import java.util.Properties;public class FlinkKafkaSimpleSchema {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();/*** 设置检查点*/env.enableCheckpointing(5000);ObjectMapper mapper = new ObjectMapper();try {/*** 加载外部规则*/GroovyClassLoader loader = new GroovyClassLoader();File file = new File("./src/main/java/cepengine/scripts/sRule.groovy");Class aClass = loader.parseClass(file);GroovyObject groovyObject = (GroovyObject) aClass.newInstance();Pattern<Event, Event> pattern = (Pattern<Event, Event>) groovyObject.invokeMethod("run", null);/** 初始化 Consumer 配置 */Properties consumerConfig = new Properties();consumerConfig.setProperty("bootstrap.servers", "localhost:9092");consumerConfig.setProperty("group.id", "risk_control");/** 初始化 Kafka Consumer */FlinkKafkaConsumer<Event> flinkKafkaConsumer =new FlinkKafkaConsumer<Event>("flink_kafka_poc_input",new InputEventSchema(),consumerConfig);/** 配置offset */flinkKafkaConsumer.setStartFromEarliest();/** 将 Kafka Consumer 加入到流处理 */DataStream<Event> stream = env.addSource(flinkKafkaConsumer);/*** 匹配规则*/PatternStream<Event> patternStream = CEP.pattern(stream, pattern);DataStream<Event> outstream = patternStream.select(new PatternSelectFunction<Event, Event>() {@Overridepublic Event select(Map<String, List<Event>> map) throws Exception {List<Event> next = map.get("next");return new Event(next.get(0).getKey(), next.get(0).getValue(), next.get(0).getTopic(),next.get(0).getPartition(),next.get(0).getOffset());}});outstream.print("next");/** 初始化 Producer 配置 */Properties producerConfig = new Properties();producerConfig.setProperty("bootstrap.servers", "localhost:9092");producerConfig.setProperty("max.request.size", "102428800");/** 初始化 Kafka Producer */FlinkKafkaProducer<Event> myProducer = new FlinkKafkaProducer<Event>("flink_kafka_poc_output",new OutputEventSchema(),producerConfig);/** 将 Kafka Producer 加入到流处理 */outstream.addSink(myProducer);/** 执行 */env.execute();} catch (Exception e) {}}}

 

Groovy脚本：

package cepengine.scriptsimport cepengine.domain.Event
import org.apache.flink.cep.pattern.Pattern
import org.apache.flink.cep.pattern.conditions.SimpleCondition
import org.apache.flink.streaming.api.windowing.time.Timeclass sRule implements Serializable{def run() {Pattern<Event, ?> pattern =Pattern.<String>begin("begin").where(new SimpleCondition<Event>() {@Overrideboolean filter(Event event) throws Exception {return  event.getValue().contains("失败")}}).next("next").where(new SimpleCondition<Event>() {@Overrideboolean filter(Event event) throws Exception {return event.getValue().contains("失败")}})//
//                .next("next2")
//                .where(new SimpleCondition<LoginEvent>() {
//            @Override
//            boolean filter(LoginEvent loginEvent) throws Exception {
//                return loginEvent.getType().equals("success")
//            }
//        }).within(Time.seconds(5))return pattern}
}

kafka consumer schema:

package cepengine.domain;import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;import java.io.IOException;
import java.nio.charset.StandardCharsets;import static org.apache.flink.api.java.typeutils.TypeExtractor.getForClass;public class InputEventSchema implements KeyedDeserializationSchema<Event> {@Overridepublic Event deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {String msg = new String(message, StandardCharsets.UTF_8);String key = null;if (messageKey != null) {key = new String(messageKey, StandardCharsets.UTF_8);}return new Event( key, msg,topic, partition, offset);}@Overridepublic boolean isEndOfStream(Event nextElement) {return false;}@Overridepublic TypeInformation<Event> getProducedType() {return getForClass(Event.class);}}

kafka producer schema:

package cepengine.domain;import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;public class OutputEventSchema implements KeyedSerializationSchema<Event> {@Overridepublic byte[] serializeKey(Event event) {return event.getKey().getBytes();}@Overridepublic byte[] serializeValue(Event event) {return event.getValue().getBytes();}@Overridepublic String getTargetTopic(Event event) {return null;}}

Event类：

package cepengine.domain;public class Event {private String topic;private int partition;private long offset;private String value;private  String key;@Overridepublic String toString() {return "Event{" +"topic='" + topic + '\'' +", partition=" + partition +", offset=" + offset +", value='" + value + '\'' +", key='" + key + '\'' +'}';}public Event() {}public Event(String key, String value, String topic, int partition, long offset) {this.key = key;this.value = value;this.topic = topic;this.partition = partition;this.offset = offset;}public String getKey() {return key;}public void setKey(String key) {this.key = key;}public String getTopic() {return topic;}public void setTopic(String topic) {this.topic = topic;}public int getPartition() {return partition;}public void setPartition(int partition) {this.partition = partition;}public long getOffset() {return offset;}public void setOffset(long offset) {this.offset = offset;}public String getValue() {return value;}public void setValue(String Value) {this.value = value;}}

目前只是基于文件加载规则，后续改造为基于数据库加载规则。

这篇关于flink 加载外部cep规则的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/240429。

相关文章

Flutter 进阶：绘制加载动画

Flutter 进阶：绘制加载动画

绘制加载动画：由小圆组成的大圆 1. 定义 LoadingScreen 类2. 实现 _LoadingScreenState 类3. 定义 LoadingPainter 类4. 总结 实现加载动画 我们需要定义两个类：LoadingScreen 和 LoadingPainter。LoadingScreen 负责控制动画的状态，而 LoadingPainter 则负责绘制动画。
阅读更多...
Adblock Plus官方规则Easylist China说明与反馈贴（2015.12.15）

Adblock Plus官方规则Easylist China说明与反馈贴（2015.12.15）

-------------------------------特别说明--------------------------------------- 视频广告问题：因Adblock Plus的局限，存在以下现象，优酷、搜狐、17173黑屏并倒数；乐视、爱奇艺播放广告。因为这些视频网站的Flash播放器被植入了检测代码，而Adblock Plus无法修改播放器。 如需同时使用ads
阅读更多...
Flink任务重启策略

Flink任务重启策略

概述 Flink支持不同的重启策略，以在故障发生时控制作业如何重启集群在启动时会伴随一个默认的重启策略，在没有定义具体重启策略时会使用该默认策略。如果在工作提交时指定了一个重启策略，该策略会覆盖集群的默认策略默认的重启策略可以通过 Flink 的配置文件 flink-conf.yaml 指定。配置参数 restart-strategy 定义了哪个策略被使用。常用的重启策略： 固定间隔 (Fixe
阅读更多...
使用WebP解决网站加载速度问题，这些细节你需要了解

使用WebP解决网站加载速度问题，这些细节你需要了解

说到网页的图片格式，大家最常想到的可能是JPEG、PNG，毕竟这些老牌格式陪伴我们这么多年。然而，近几年，有一个格式悄悄崭露头角，那就是WebP。很多人可能听说过，但到底它好在哪？你的网站或者项目是不是也应该用WebP呢？别着急，今天咱们就来好好聊聊WebP这个图片格式的前世今生，以及它值不值得你花时间去用。 为什么会有WebP？ 你有没有遇到过这样的情况？网页加载特别慢，尤其是那
阅读更多...
gazebo 已加载模型但无法显示

gazebo 已加载模型但无法显示

目录 写在前面的话问题一：robot_state_publisher 发布机器人信息失败报错一 Error: Error document empty.报错二 .xcaro 文件中有多行注释成功启动 问题二：通过 ros2 启动 gazebo 失败成功启动 问题三：gazebo 崩溃和无法显示模型问题四： 缺少 robot_description 等话题正确的输出 写在前面的话
阅读更多...
关联规则（一）Apriori算法

关联规则（一）Apriori算法

此篇文章转自 http://blog.sina.com.cn/s/blog_6a17628d0100v83b.html 个人觉得比课本上讲的更通俗易懂！ 1．  挖掘关联规则 1.1   什么是关联规则 一言蔽之，关联规则是形如X→Y的蕴涵式，表示通过X可以推导“得到”Y，其中X和Y分别称为关联规则的先导(antecedent或left-hand-side, LHS)和后
阅读更多...
JVM类的加载器及加载过程

JVM类的加载器及加载过程

类的加载器及加载过程 文章目录 类的加载器及加载过程类的加载过程加载：链接(验证、准备、解析)：初始化： 类加载器的分类引导类加载器：BootstrapClassLoader 启动类加载器（ C/C++实现，嵌套在JVM内部）自定义类加载器（所有派生于抽象类ClassLoader的类加载器）获取ClassLoader的途径 双亲委派机制（重点）判断两个Class对象是否为同一个类
阅读更多...
电子电气架构---私有总线通信和诊断规则

电子电气架构---私有总线通信和诊断规则

电子电气架构—私有总线通信和诊断规则 我是穿拖鞋的汉子，魔都中坚持长期主义的汽车电子工程师。 老规矩，分享一段喜欢的文字，避免自己成为高知识低文化的工程师： 屏蔽力是信息过载时代一个人的特殊竞争力，任何消耗你的人和事，多看一眼都是你的不对。非必要不费力证明自己，无利益不试图说服别人，是精神上的节能减排。 无人问津也好,技不如人也罢,你都要试着安静下来,去做自己该做的事.而不是让内心的烦躁、
阅读更多...
Unity Adressables 使用说明（六）加载（Load） Addressable Assets

Unity Adressables 使用说明（六）加载（Load） Addressable Assets

【概述】Load Addressable Assets Addressables类提供了加载 Addressable assets 的方法。你可以一次加载一个资源或批量加载资源。为了识别要加载的资源，你需要向加载方法传递一个键或键列表。键可以是以下对象之一： Address：包含你分配给资源的地址的字符串。Label：包含分配给一个或多个资源的标签的字符串。AssetReference Obj
阅读更多...
Eclipse发布Maven项目到tomcat，无法加载到lib文件夹下的jar包

Eclipse发布Maven项目到tomcat，无法加载到lib文件夹下的jar包

BMS 解决方法： 当我们发布web项目到tomcat时，访问地址时会报一个classnotfound的错误，但是eclipse中的项目中都已经添加了相应的类，有一种比较容易犯的错误是，你没有把额外所需的jar包加到tomcat中的lib文件夹中，在这里介绍一种在项目中直接添加jar包到lib目录下：  右键已创建的web项目——properties属性——点击Deployment Assem
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2