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。

相关文章

最好用的WPF加载动画功能

最好用的WPF加载动画功能

《最好用的WPF加载动画功能》当开发应用程序时,提供良好的用户体验（UX）是至关重要的,加载动画作为一种有效的沟通工具,它不仅能告知用户系统正在工作,还能够通过视觉上的吸引力来增强整体用户体验,本文给... 目录前言需求分析高级用法综合案例总结最后前言当开发应用程序时，提供良好的用户体验（UX）是至关重要
阅读更多...
MyBatis延迟加载的处理方案

MyBatis延迟加载的处理方案

《MyBatis延迟加载的处理方案》MyBatis支持延迟加载（LazyLoading）,允许在需要数据时才从数据库加载,而不是在查询结果第一次返回时就立即加载所有数据,延迟加载的核心思想是,将关联对... 目录MyBATis如何处理延迟加载？延迟加载的原理1. 开启延迟加载2. 延迟加载的配置2.1 使用
阅读更多...
Android WebView的加载超时处理方案

Android WebView的加载超时处理方案

《AndroidWebView的加载超时处理方案》在Android开发中,WebView是一个常用的组件,用于在应用中嵌入网页,然而,当网络状况不佳或页面加载过慢时,用户可能会遇到加载超时的问题,本... 目录引言一、WebView加载超时的原因二、加载超时处理方案1. 使用Handler和Timer进行超
阅读更多...
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对象是否为同一个类
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2