Flink如何使用CEP处理过期密钥
Flink是一个开源的流式处理框架，它提供了强大的流式计算能力和丰富的API，可以用于实时数据处理和分析。CEP（Complex Event Processing）是Flink的一个功能模块，用于处理复杂事件。
在Flink中使用CEP处理过期密钥的步骤如下：
- 导入依赖：首先需要在项目中导入Flink和CEP的相关依赖，可以通过Maven或Gradle进行管理。
- 定义事件模型：根据实际需求，定义过期密钥的事件模型，包括事件的属性和字段。
- 创建数据流：使用Flink的DataStream API创建输入数据流，将过期密钥的事件数据输入到流中。
- 定义模式：使用CEP的Pattern API定义过期密钥的模式，可以通过指定事件的属性和条件来匹配模式。
- 应用模式：将定义好的模式应用到输入数据流上，使用CEP的
pattern()方法进行模式匹配。 - 处理匹配结果：根据匹配到的模式，可以执行相应的处理逻辑，例如发送警报、更新数据库等。
- 启动作业：使用Flink的ExecutionEnvironment或StreamExecutionEnvironment启动作业，开始处理过期密钥的事件流。
Flink CEP的优势在于其高性能和灵活性，可以处理大规模的实时数据，并支持复杂的事件模式匹配。它适用于各种实时监控、异常检测、欺诈检测、实时报警等场景。
腾讯云提供了一系列与Flink相关的产品和服务，包括云原生数据库TDSQL、云原生数据仓库TDSW、云原生数据湖TDH、云原生数据计算TDC等。这些产品可以与Flink结合使用，提供全面的数据处理和分析解决方案。具体产品介绍和链接地址如下：
- 腾讯云云原生数据库TDSQL：提供高性能、高可用的云原生数据库服务，支持MySQL和PostgreSQL。了解更多：TDSQL产品介绍
- 腾讯云云原生数据仓库TDSW：提供海量数据存储和分析能力，支持PB级数据规模。了解更多：TDSW产品介绍
- 腾讯云云原生数据湖TDH：提供大规模数据存储和计算能力，支持数据湖架构。了解更多：TDH产品介绍
- 腾讯云云原生数据计算TDC：提供大规模数据计算和分析能力，支持Flink、Spark等计算引擎。了解更多：TDC产品介绍
通过结合Flink和腾讯云的相关产品，可以构建强大的实时数据处理和分析平台，满足各种业务需求。
相关·内容
我有流作业，监听事件，使用CEP对它们进行操作。 流程是 stream = source
.assignTimestampsAndWatermarks(...)
.filter(...);
CEP
.pattern(stream.keysBy(e-> e.getId()), pattern)
.process(PattenMatchProcessFunction)
.sink(...); 所有的键都是短暂的，并且进程函数不包含任何状态，也就是说，可以通过设置ttl来删除状态。使用EventTime特性 我的问题是，fli
浏览 20提问于2021-01-20得票数 0
1回答
我是Flink和Flink CEP的新手。使用事件时间时，Flink CEP是否以升序(事件-)时间顺序迭代事件以检查模式？提前感谢！
浏览 33提问于2019-02-28得票数 0
回答已采纳
我们的Flink工作遇到了一个很难观察的问题.
这份工作相当简单，它：
使用Flink Kinesis连接器从Kinesis读取消息
键入消息，并将它们分发给~30个不同的WindowFunctions操作符，外加几个自定义的CEP操作符
从CEP/Windows发出的消息转发给将消息写入SQS的SinkFunction
我们运行的是Flink 1.10.1 Fargate，使用的是2个4vCPU/8GB的容器，我们使用的是具有以下配置的RocksDB状态后端：
state.backend: rocksdb
state.backend.async: true
state.b
浏览 14提问于2020-10-01得票数 0
1回答
我有两个来自两个Kafka主题的流表，我想加入这些流并对所连接的数据执行聚合功能。需要使用滑动窗口连接流。在连接和窗口数据时，我将得到一个错误Rowtime attributes must not be in the input rows of a regular join. As a workaround you can cast the time attributes of input tables to TIMESTAMP before.。
下面是代码片段
select cep.payload['id'] , ep.payload['id'] ,
浏览 25提问于2022-05-24得票数 0
我是新的Flink CEP，并试图测试基本的东西-在下面的代码，我的期望是所有的输入应该匹配模式，并应打印为匹配的结果。但不知何故，没有什么比('matechedStream.print()')更能说明原因吗？
任何建议/帮助都将不胜感激。
package com.o9.flink;
import com.o9.flink.asyncio.DemandSupply;
import org.apache.flink.cep.CEP;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
浏览 9提问于2022-08-09得票数 0
回答已采纳
2回答
我有一个从套接字读取并检测模式的flink cep代码。让我们说这个模式(单词)是“警觉”的。如果警告一词出现五次或更多次，则应创建警报。但是我得到了一个输入不匹配错误。Flink版本为1.3.0。提前谢谢!！
package pattern;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.Iterativ
浏览 3提问于2017-07-11得票数 1
回答已采纳
我开始研究用Scala语言编写的Apache Flink的CEP库，当我试图通过执行https://ci.apache.org/projects/flink/flink-docs-stable/dev/libs/cep.html教程中所示的CEP.pattern(input,pattern)来创建一个PatternStream时，集成开发环境说它“无法解析重载方法”，它指的是pattern方法。根据我用来分别创建输入和模式的readTextFile和Pattern[String].begin('line').where(_.length == 10)的实现，方法的参数或泛型类
浏览 50提问于2020-01-14得票数 0
回答已采纳
2回答
我正在尝试运行这个程序: Flink CEP监控
1:在亚马逊的上安装了openjdk1.8和Flink 1.0.2。但是，当我试图运行这个程序时，它会抛出以下异常：
缺少“Map”的泛型类型参数。您的编译器似乎没有将它们存储到.class文件中。目前，只有Eclipse编译器保存使用lambdas特性类型所需的类型信息--安全。有关如何编译包含lambda表达式的作业的详细信息，请参阅文档。org.apache.flink.api.java.typeutils.TypeExtractor.validateLambdaGenericParameter(TypeExtractor.java:1
浏览 3提问于2016-05-05得票数 2
1回答
我想检测所有相同整数的序列，后面跟着这些整数的计数。有什么方法可以用Flink CEP库创建这个模式吗？假设如下：
val input List(1,2,2,2,3,3,2,1,1,1,3,0,1)
这些序列应该被检测到：
List(2,2,2,3)
List(3,3,2)
List(1,1,1)
我知道有很多方法可以通过Flink本身来完成它，但是我想用CEP来完成它，我认为它需要支持将上下文传递到下一个条件，对吗？如果是这样的话，Flink CEP是否可以传递上下文或修改流中的事件，并将一些额外的数据传递给下一个条件？
浏览 6提问于2022-09-10得票数 0
回答已采纳
我用Flink实现了一个模式，它匹配三个事件，比如A->B->C。在定义了我的模式之后，我生成一个
PatternStream<Event> patternStream = CEP.pattern(eventStream, pattern);
有了这样一个PatternSelectFunction
patternStream.select(new MyPatternSelectFunction()).print();
这就像一种魅力，但我对所有匹配事件的活动时间感兴趣。我知道传统的Flink流API提供了丰富的功能，允许您注册Flink的内部延迟跟踪器，如本所述。我还看
浏览 0提问于2019-01-21得票数 1
回答已采纳
问题1：我正在研究一个案例场景，我们需要融合来自多个传感器的数据，例如8个传感器，并以树的形式将它们连接起来。例如，加入s1、s2、s3 s4来形成流A，然后加入s5、s6、s7和s8来形成B流，然后在流A和B上执行CEP。
问题2 :可以在多个流上执行CEP吗？中清楚地提到，模式将应用于一个流。
DataStream<Event> input = ...
Pattern<Event, ?> pattern = ...
PatternStream<Event> patternStream = CEP.pattern(input, pattern);
如果不
浏览 4提问于2017-10-29得票数 2
回答已采纳
我写了一个模式。我有一个条件列表(来自json的获取规则).Data(Json)来自kafka服务器。我想用这个列表过滤数据。但它不起作用。我该怎么做呢？对于keyedstream和警报，我不是很确定。flink能像这样工作吗？
主程序：
package cep_kafka_eample.cep_kafka;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import c
浏览 1提问于2017-09-21得票数 0
我正在尝试在无序流事件的FLINK上实现CEP模式。我的溪流就是这样构建的：
DataStream<DataInput> input = inputStream.flatMap(
new FlatMapFunction<String, DataInput>() {
@Override
public void flatMap(String value, Collector<DataInput> out) throws Exception {
for(DataInput
浏览 11提问于2022-01-28得票数 0
我遵循了，并用kafka json实现了相同的样本数据。
消费者样本数据{"temperature" : 28,"machineName":"xyz"}
DataStream<Alert> patternStream = CEP.pattern(inputEventStream, warningPattern)
.flatSelect(new PatternFlatSelectFunction<TemperatureEvent, Alert>() {
private static f
浏览 0提问于2019-02-27得票数 0
我正在制作一个Flink CEP应用程序，通过Kafka读取数据。当我试图捕捉模式时，当它之后没有数据时，接收器操作就不会发生。例如，我希望A-> B-> C作为一个模式。卡夫卡提供了A，B，C数据。但是，为了使我添加到patternProcess函数中的接收器操作工作，来自kafka的数据必须类似于A、B、C、X。如何解决这个问题请帮助。
读KAFKA
DataStream<String> dataStream = env.addSource(KAFKA).assignTimestampsAndWatermarks(WatermarkStrategy
浏览 1提问于2021-05-21得票数 0
我试图创建一个简单的应用程序，应用程序将使用Kafka消息，做一些cql转换并发布到Kafka，下面是代码：
JAVA: 1.8 Flink: 1.13 Scala: 2.11 flink-siddhi: 2.11-0.2.2-快照
我使用的是库：
卡夫卡的输入json：
{
"awsS3":{
"ResourceType":"aws.S3",
"Details":{
"Name":"crossplane-test",
"Cr
浏览 6提问于2021-05-23得票数 0
回答已采纳
这就是我尝试初始化StreamTableEnvironment时遇到的错误
private final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
Exception in thread "main" java.lang.NoSuchMethodError: org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getConfiguration()Lorg/apache/flink/configur
浏览 6提问于2022-05-30得票数 0
1回答
我正在尝试打印一个字符串，如果使用Flink CEP库找到Hello和world。我的来源是Kafka，并使用控制台生产者输入数据。这一部分正在发挥作用。我可以打印出我在主题中输入的内容。但是，它不会打印出我的最后一条消息“世界真好！”。它甚至不会打印出它进入了lambda。下面是类
package kafka;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.fl
浏览 10提问于2016-09-19得票数 2
在engine事件模式中，有Every操作符，它指示当每个关键字限定的子表达式计算为true或false时，模式子表达式应该重新启动。
，我们可以以不同的方式使用every操作符，例如：
A -> B
every ( A -> B )
every A -> B
A -> every B
every A -> every B
我想知道在Flink CEP中哪些模式是可能的？我想在Flink中使用every A -> B，但不知道如何使用。
以上哪种模式-- Flink中的默认CEP模式--代表了哪种模式？(使用Flink
浏览 0提问于2017-01-09得票数 0
回答已采纳
我使用的是使用Java的Flink表API，其中我想将DataSet转换为DataStream .以下是我的代码：
TableEnvironment tableEnvironment=new TableEnvironment();
Table tab1=table.where("related_value < 2014").select("related_value,ref_id");
DataSet<MyClass>ds2=tableEnvironment.toDataSet(tab1, MyClass.class);
DataStream&
浏览 4提问于2016-05-13得票数 4
我的饮料的内容是
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses
浏览 1提问于2017-11-09得票数 0
回答已采纳
我是Flink的新手，我正在尝试POC，在该POC中，如果在超过CEP中指定的时间段内的x时间量内未收到事件 public class MyCEPApplication {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
streamExecutionEnvi
浏览 20提问于2020-10-18得票数 0
回答已采纳
我的代码如下：
import java.util.Properties
import com.google.gson._
import com.typesafe.config.ConfigFactory
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.cep.scala.CEP
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import
浏览 1提问于2017-07-07得票数 1
回答已采纳
我想做Flink CEP引擎的性能分析，我遇到了这些类。
org.apache.flink.optimizer.costs.CostEstimator;
org.apache.flink.optimizer.costs.Costs;
org.apache.flink.optimizer.costs.DefaultCostEstimator;
但问题是，我不知道如何使用这两个类。有人能给我提供一个代码或暗示，我如何才能找到运算符{ join例如Flink }的成本估算。
下面是我在Flink中执行的联接的代码
DataStream<JoinedEvent> joinedEvent
浏览 5提问于2017-10-23得票数 0
回答已采纳
我刚从Flink CEP开始，我来自Esper CEP引擎。您可能知道(或不知道)，在Esper中使用它们的语法(EPL)，您可以轻松地创建一个batch或slide窗口，在这些窗口中对事件进行分组，并允许您将这些事件与函数(avg、max、min、.)一起使用。
例如，使用下面的模式，您可以创建一个5秒的批处理窗口，并计算在该指定窗口中接收到的所有Stock事件的属性Stock的平均值。
select avg(price) from Stock#time_batch(5 sec)
问题是，我想知道如何在Flink CEP上实现这一点。我知道，Flink CEP中的目标或方法可能是不同的，因此
浏览 0提问于2018-10-05得票数 0
回答已采纳
我试着理解如何在Flink工程中CEP。例如，来自pages：和。
我想知道Flink CEP应用程序中(Scala)代码的基本结构是什么？
应该使用哪些(必需的)部件、类、函数/方法等，以及按什么顺序使用？
我认为基本部分可以是: 1.模式定义2.模式流定义(或者可以使用许多并行流吗?) 3.选择定义是正确的吗？
如何使用"select part“以及如何查找有关它的更多信息？它看起来相当复杂。
浏览 0提问于2018-03-06得票数 0
在过去几天学习Flink CEP库时，我的印象是它没有为Flink的标准功能添加任何新的基本功能。似乎Flink CEP的唯一目的是使事件处理更容易，具有清晰的语义和直观的代码结构。例如，Flink只显示事件匹配跳过的。虽然这些语义对于很多情况来说都是足够的，但它可能不能解决具体的问题，这使我们回到了普通的Flink。
测试用例是以下模式：
Emmit a alert(represented by 'a') for each non-overlapping pair of numbers in a stream
以模式为代表：
Pattern.begin[EventType](
浏览 6提问于2020-03-05得票数 0
回答已采纳
我试图发送一个警报，每次发生的事件具有相同的ID。为此，我使用keyBy()应用我的操作根据ID。
import java.util.Properties
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.{TimeCharacteristic, scala}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache
浏览 3提问于2020-03-02得票数 0
回答已采纳
有人能帮我解决以下错误吗？我认为版本有一些问题，但我不知道如何修复它。我试图运行Apache，我使用Kafka主题作为输入，在尝试运行Flink时，我得到了以下错误：
错误：
线程"main“中的异常:org/apache/flink/util/ org.apache.flink.cep.pattern.Pattern.where(Pattern.java:130) at hes.cs63.CEPMonitor.CEPMonitoring.main(CEPMonitoring.java:55)，由: java.lang.ClassNotFoundException: org.apac
浏览 3提问于2017-11-20得票数 1
我正在我的Flink 1.11应用程序中创建一个StreamTableEnvironment，这在Windows中有效，但是，当我试图在Debian中使用JDK9(在Windows中是相同版本)部署mi jar时，它抛出了这个错误： Exception in thread "main" java.lang.ExceptionInInitializerError
at org.apache.flink.table.planner.calcite.FlinkRelFactories$.<init>(FlinkRelFactories.scala:51)
浏览 38提问于2020-10-27得票数 0
回答已采纳
flink运行/home/admin/Documents/flink_cep/Flink-master/dist/Kinesis.jar
当我试图在命令行中运行Jar文件时，会出现错误，但我的代码在Netbeans IDE中运行得很好：
浏览 0提问于2018-08-17得票数 0
回答已采纳
我正在寻找一个可视化的编程工具，以抽象化的代码，当创建数据流流，以建立在Flink或束-Flink中的CEP应用程序。类似于Nifi或流设置数据收集器的图形设计器。
有人知道这样的计划吗？
浏览 5提问于2017-04-30得票数 4
回答已采纳
1回答
我在Flink SQL中使用CEP模式，它正在按预期工作，连接到Kafka broker。但是当我连接到基于集群的云kafka设置时，Flink CEP并没有触发。以下是我的sql：
create table agent_action_detail
(
agent_id String,
room_id String,
create_time Bigint,
call_type String,
application_id String,
connect_time Bigint,
row_time TIMESTAMP_LTZ(
浏览 1提问于2021-10-25得票数 0
回答已采纳
我正在使用Flink CEP模块，我想知道是否将一个函数传递给where子句，它将返回布尔值，它是否会以分布式方式工作。
示例-：val pattern= Pattern.start("begin").where(v=>booleanReturningFunction(v))
上述代码在以简单条件作为CEP的flink作业提交时，会以分布式方式工作吗？
浏览 6提问于2020-04-19得票数 0
回答已采纳
1回答
我对在Flink中可以并行工作的任务的数量感到困惑，
谁能跟我解释一下：
在分布式系统中的并行性是多少？以及它与Flink中的computation?Is术语的关系，它是否与我们所说的2并行=2任务并行工作相同? Flink中的，如果两个操作符分别工作，但每一个操作符中的并行数为1，这是否算作并行？在KeyedStream中，最大并行度是键数吗？，Flink中当前的CEP引擎是否能够在多个任务中工作？
谢谢。
浏览 4提问于2020-04-10得票数 2
我使用flink cep sql和闪烁计划器。这是我的sql
select * from test_table match_recognize (
partition by agent_id,room_id,call_type
order by row_time // process time
measures
last(BF.create_time) as create_time,
last(AF.connect_time) as connect_time
one row per match after match SKIP PA
浏览 3提问于2021-12-08得票数 0
回答已采纳
假设我有两种不同类型的数据流，一种提供天气数据，另一种提供车辆数据，我想使用Flink对数据进行复杂的事件处理。
Flink 1.3.x中的哪种方法是正确的方法？我看到了不同的方法，如Union、Connect、Window Join。基本上，我只想尝试这样一个简单的CEP：
IF weather is wet AND vehicle speed > 60
WITHIN the last 10 seconds
THEN raise alert
谢谢!
浏览 3提问于2017-07-12得票数 3
回答已采纳
在Apache Flink中获取Exception in thread "main" java.lang.IndexOutOfBoundsException。我试着在效仿，但我得到了免责的提法。下面是我的主类的一个片段。我将感谢任何关于我做错了什么或如何排除错误的指针。
DataStream<CSBEvent> inputEventStream = env.addSource(
new FlinkKafkaConsumer09<CSBEvent>("tester03", new EventDeseriali
浏览 1提问于2017-11-24得票数 0
在我的flink作业中，我将数据保存在mapstate中几分钟，以便进一步处理，在process元素中，我需要在mapstate值的帮助下执行一组操作，所以我在mapstate内容的帮助下使用多个线程进行处理。线程没有修改状态，它只获取导致CocurrentModification异常的键值，这个TTL异常也出现在COnfigured TTL时间之前，日志如下。 `java.util.concurrent.ExecutionException: org.apache.flink.util.FlinkRuntimeException: Failed to incrementally clean
浏览 84提问于2020-04-03得票数 0
1回答
Flink和动态模板识别
、、、
我们计划使用Flink CEP根据一些动态模板处理大量事件。系统必须识别事件链(有时是带有条件和分组的复杂链)。模板将由用户创建。换句话说，我们必须在不接触代码的情况下创建复杂的模板。是否可以使用Apache Flink来解决此问题？Filnk支持动态模板吗？
浏览 3提问于2016-10-13得票数 1
如何使用flink cep库检测胡言乱语模式？
例如:假设设备有一些问题，那么它会不断地发布像开、关这样的值。如果问题存在30分钟，如何使用CEP检测模式。下面是我提到的一些样本数据。
OFF 16/08/18 11:38
ON 16/08/18 11:38
OFF 16/08/18 11:38
ON 16/08/18 11:37
OFF 16/08/18 11:37
ON 16/08/18 11:36
OFF 16/08/18 11:36
OFF 16/08/18 11:36
ON 16/08/18 11:36
浏览 0提问于2020-05-15得票数 1
1回答
FlinkCEP的执行时间
、、、、
我使用Flink和FlinkCEP来检测数据流上的复杂事件。出于研究目的，我只需要测量识别时间。 我正在使用Flink / FlinkCEP - 1.7.1。我使用env.fromCollection()函数在Flink环境中创建流。在此之后，我将使用FlinkCEP：CEP.pattern(....)以及其他select和print函数。 我只找到了这篇文章：Measure job execution time in flink，它帮了我很多。它提出了一种返回流环境进程的执行时间的解决方案。这不完全是我要找的。 我注意到返回值包含了其他运算符的时间，比如.assignAscendingTi
浏览 26提问于2019-09-04得票数 1
在我们的应用程序中，flink检查点的大小正在增加，并且永远不会以rocks db作为状态后端。(AWS KDA)
Kafka-->do some magic -->ES(Sink)-->Writes to kafka
这里我们使用的键是UUID，并且不会重复。如何配置以确保检查点大小不会增加，或者对db进行微调以删除少于1天的任何较旧的密钥。
浏览 3提问于2021-03-01得票数 0
1回答
我正在尝试了解在提交Flink作业之前需要考虑的重要功能有哪些。
我的问题是并行的数量是多少，(物理上)有上限吗？并行性如何影响我的工作表现？
例如，我有一个从无键数据流中检测模式的CEP Flink作业，除非我使用KeyBy操作符对数据流进行分区，否则并行度的数量将始终为1。
如果我错了，请纠正我：
如果我对数据流进行分区，那么我的并行度数量将等于不同键的数量。但问题是，模式匹配是针对每个键独立完成的，因此我不能定义需要来自具有不同键的两个分区的信息的模式。
浏览 0提问于2020-05-15得票数 0
我的问题是，如果我们有两个原始事件流，即烟雾流和温度流，并且我们想知道是否有复杂的事件，即Fire，通过将操作符应用于原始流，我们能在Flink中这样做吗？
我之所以问这个问题，是因为到目前为止，我所看到的Flink CEP的所有例子都只有一个输入流。如果我错了，请纠正我。
浏览 0提问于2017-07-28得票数 2
回答已采纳
2回答
下面显示了在Flink-CEP中检测模式的代码
// Generate temperature warnings for each matched warning pattern
DataStream<TemperatureEvent> warnings = tempPatternStream.select(
(Map<String, MonitoringEvent> pattern) -> {
TemperatureEvent first = (TemperatureEvent) pattern.get("first")
浏览 0提问于2017-08-13得票数 3
回答已采纳
这一次我的问题是:使用MapState，使用mapstate.put(key, value)将键的当前值修改到mapState中是安全的，或者我需要执行mapState.remove(key)，然后再次执行mapstate.put(key, value)，或者无论如何都要更新此值？ 从Flink的状态抽象开始，抽象不是为并发访问而设计的，也不应该在多个线程之间共享。那么，重新表述我的问题:我可以根据键将值更新到mapState中，而不需要删除键然后再放入键吗？如何避免使用mapState的ConcurrentModificationException，而不将该操作符的并行度设置为1？ 因为我
浏览 43提问于2020-04-16得票数 1
1回答
请帮帮我，我有两个问题：
我从Apache消息(然后我有步骤:反序列化到POJO，过滤器，keyBy .)
最好使用的是: KeyedProcessFunction (带有状态、定时器、if-逻辑块)或Flink CEP模式库？。
我可以检查KeyedProcessFunction中的输入序列(检查状态、if- state块、out.collect(.)、state.clear()...you将理解我)，也可以使用条件和量化器使用Flink CEP库。
如何停止flink CEP模式？
例如：
我有输入序列: A1，(无事件1分钟) A2，(无事件5分钟)А3，(无事件1分钟)А4，(无事件超
浏览 4提问于2020-08-13得票数 1
回答已采纳
1回答
Flink CEP如何管理间歇性状态？它将它们存储在哪里？它只是在内存中，还是有一个快速的持久存储来支持这些状态？
文档没有在任何地方提到这一点。
浏览 13提问于2017-12-16得票数 3
回答已采纳
1回答
有没有办法以可共享的格式(例如Json)指定Apache Flink CEP模式？
我的用例是:开发一个规则引擎，允许用户在不编写大量Java代码的情况下定制CEP模式，并轻松地与其他人共享模式。
浏览 2提问于2018-12-11得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
