在Flink中停用KeyedProcessFunction上的onTimer
是指在使用KeyedProcessFunction时，禁用onTimer方法的调用。KeyedProcessFunction是Flink中用于处理KeyedStream的函数，它可以访问事件流中的每个元素，并根据键（key）对元素进行处理。
onTimer方法是KeyedProcessFunction的一个回调方法，用于处理定时器事件。定时器事件是在指定的时间点触发的事件，可以用于执行一些延迟操作或定期任务。在onTimer方法中，可以执行一些自定义的逻辑，例如发送通知、更新状态等。
然而，有时候我们可能不需要使用定时器功能，或者希望在某些情况下禁用定时器。在这种情况下，可以通过不在KeyedProcessFunction中实现onTimer方法来停用定时器。
停用KeyedProcessFunction上的onTimer的步骤如下：
- 创建一个继承自KeyedProcessFunction的自定义函数类，例如MyProcessFunction。
- 在自定义函数类中，只实现processElement方法，不实现onTimer方法。processElement方法用于处理每个输入元素。
- 在processElement方法中，可以根据需要执行一些逻辑，例如过滤、转换、聚合等。
- 将自定义函数类应用于KeyedStream，例如：
- 将自定义函数类应用于KeyedStream，例如：
- 这里假设DataStreamElement是输入数据的类型，ResultElement是处理结果的类型，getKey是获取元素键的方法。
通过以上步骤，就可以在Flink中停用KeyedProcessFunction上的onTimer方法。这样，定时器事件将不会被触发，只会执行processElement方法中的逻辑。
需要注意的是，停用定时器可能会影响某些需要定时触发的功能，因此在实际应用中需要根据具体需求进行判断和决策。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云Flink产品：https://cloud.tencent.com/product/flink
相关·内容
我有一个带有数据和时间戳的记录日志，我的Flink应用程序按照上升时间戳的顺序接收这些记录。在某个键的第一项到达窗口后，我希望在X事件发生后关闭窗口，并检查是否有足够的项到达某个条件，并为该键发出pass或fail。
Flink中的基本窗口函数是不可能的吗？例如，如果我希望我的窗口有30秒长，但是键中的第一个项到达15秒，最后一个项到达40秒，则该窗口似乎将在30秒关闭，并且该键的记录跟踪将被分割为两个窗口。在这种情况下，我想要的是窗口在事件发生时从15秒开始，在45秒关闭。
浏览 0提问于2019-07-16得票数 0
回答已采纳
我已经在一个流中输入了关键事件，我希望按键累积，直到超时(例如，5分钟)，然后处理累积到那个点的事件(然后忽略该键之后的所有内容，但首先是第一件事)。
我对Flink并不熟悉，但在概念上，我认为我需要类似下面的代码。
DataStream<Tuple2<String, String>> dataStream = see
.socketTextStream("localhost", 9999)
.flatMap(new Splitter())
.keyBy(0)
浏览 0提问于2018-12-05得票数 1
回答已采纳
在引用Flink代码示例时，我观察到当使用connect操作符连接两个流时，coProcessFunction和keyedCoProcessFunction在操作键控流(覆盖processElement1、processElement2和onTimer)时几乎可以互换和相同地扩展，所以当操作键控流时，与keyedCoProcessFunction相比，扩展CoProcessFunction来实现键控连接流的业务逻辑有什么不同？ 谢谢。
浏览 127提问于2020-09-02得票数 1
回答已采纳
假设Kafka消息包含flink窗口大小配置。
我想阅读卡夫卡的信息，并在flink中创建一个全球窗口。
问题陈述：
我们能使用BroadcastStream来处理上述场景吗？
或
是否有其他方法支持上述情况？
浏览 0提问于2020-01-20得票数 1
我对Flink很陌生，并试图了解Flink命令如何在并行性下的processElement()抽象中调用KeyedProcessFunction。考虑这个产生部分和流的例子：
package sample
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.{DataSt
浏览 2提问于2021-04-14得票数 1
回答已采纳
我需要将一些配置传递给一个定制的KeyedProcessFunction，我想用KeyedOneInputStreamOperatorTestHarness测试它。处理这件事的正确方法是什么？
我已经查看了文档，但是在搜索配置时，结果指向flink中现有的信任项列表。
谢谢
浏览 1提问于2022-01-20得票数 0
在浏览Flink官方文档时，我遇到了CheckpointedFunction。想知道你为什么以及什么时候会使用这个函数。我目前正在做一个有状态的Flink作业，它在很大程度上依赖于ProcessFunction来在RocksDB中保存状态。只是想知道CheckpointedFunction是否比ProcessFunction更好。
浏览 109提问于2021-06-08得票数 2
我有一个Flink KeyedCoProcessFunction，它在一个更大的Flink流作业中注册处理时间计时器，并且我正在尝试使用Flink 为整个作业创建单元测试。但是我不能让KeyedCoProcessFunction中的onTimer()回调触发。
有没有人把这个弄好了？它需要特殊的配置吗？
切换到事件时间可以很好地工作，所以我想知道这是不是对Flink MiniCluster不起作用，或者我的实现有什么问题。
我用Scala写了一个简单的测试，看看能不能让它正常工作。
import org.apache.flink.api.common.typeinfo.TypeInformat
浏览 0提问于2020-08-26得票数 3
Flink的KeyedStreams上的fold函数已弃用。aggregate函数是私有的，而且仅限于一些预定义的聚合类型。
在WindowedStreams上有一个很好的聚合实现。所以我能做的就是定义一个带有某种AlwaysOnElementTrigger的GlobalWindow (它在每个元素上触发)。不知何故，这让我觉得不对劲，但我看不出有什么缺点。
我的问题是:我是必须接受这个弃用警告，还是GlobalWindow的变通方法有点难看，但我想要的是什么？
浏览 13提问于2019-07-12得票数 1
回答已采纳
为了提高数据处理的性能，我们将事件存储到映射中，直到事件计数达到100时才进行处理。同时，在open方法中启动一个计时器，以便每60秒处理一次数据。
当flink版本为1.11.3时，
将flink版本升级到1.13.0后，
我发现，有时事件是从卡夫卡不断消耗，但没有在RichFlatMapFunction处理，这意味着数据丢失。在重新启动服务之后，它运行良好，但几个小时后，同样的事情又发生了。
这个flink版本有什么已知的问题吗？如有任何建议，敬请见谅。
public class MyJob {
public static void main(String[] args) thro
浏览 7提问于2021-06-24得票数 1
回答已采纳
我有一个flink数据流，我正在使用KeyedProcessFunction对其进行一些处理，然后我需要在处理后将数据保存在MongoDB中，但我无法这样做。谁能给我一些关于如何将数据从flink数据流保存到Mongo的输入？
浏览 19提问于2020-12-09得票数 0
回答已采纳
我正在使用Flink 1.12并且我有一个键控流，在我的代码中看起来A和B共享相同的水印？因此，因为A的到来已经将水印升级为2020-08-30 10:50:11，所以B被确定为延迟 输出为A(2020-08-30 10:50:08, 2020-08-30 10:50:16):2020-08-30 10:50:15，没有B的输出 我想问，是否有可能使不同的密钥具有独立的水印？A的水印和B‘水印独立变化 应用程序代码为： import java.text.SimpleDateFormat
import java.util.Date
import java.util.concurrent.Tim
浏览 12提问于2021-05-02得票数 0
回答已采纳
1回答
我有一个有趣的用例，我想用Flink来测试。我有一个传入的Message流，它要么是PASS，要么是FAIL。现在，如果消息的类型是FAIL，我有一个下游ProcessFunction，它保存Message状态，然后向依赖它的一切发送pause命令。当我收到一条PASS消息，这条消息与我之前收到的FAIL相关联(通过消息id键入)，我会向之前暂停的所有内容发送resume命令。 现在，我计划使用State TTL使存储的FAIL状态过期，并在某个超时后恢复所有内容，即使我没有收到具有相同消息id的PASS消息。这可以通过单独使用Flink来完成吗?或者我是否需要一些外部计时器来向我的程序发送
浏览 11提问于2019-02-16得票数 0
回答已采纳
我正在学习Flink，我从使用DataStream进行简单的字数统计开始。为了增强处理能力，我对输出进行了过滤，只显示找到3个或3个以上单词的结果。
DataStream<Tuple2<String, Integer>> dataStream = env
.socketTextStream("localhost", 9000)
.flatMap(new Splitter())
.keyBy(0)
.timeWindow(Time.seconds(5))
浏览 190提问于2018-03-01得票数 2
回答已采纳
在Apache Flink 1.8.1上 我正在阅读https://flink.apache.org/news/2015/12/04/Introducing-windows.html，我正在尝试弄清楚使用countWindow(size: Int)是否适合我的用例:当N个用户访问我网站的帮助页面时，我想将他们都标记为客户服务扩展。我很困惑，因为关于.countWindow()的文档有限，而且我很难确认它是否已被弃用，转而采用另一种方法。 我正在尝试这个(忽略红色的高亮部分)： ? countWindow()的源代码如下所示： ? 然后，寻找文档是如此有限。在最好的情况下，我在1.3文
浏览 14提问于2020-03-31得票数 0
回答已采纳
如何为有状态进程函数创建单元测试。我有这样的东西：
private static SingleOutputStreamOperator<Tuple> methodName(KeyedStream<Event, String> stream) {
return stream.window(TumblingEventTimeWindows.of(Time.minutes(10)))
.process(new ProcessFunction());
}
和
ProcessFunction extends Process
浏览 2提问于2020-08-14得票数 2
我要求将每个键的最后25秒值保持在flink map状态中，但TTL会在达到25秒时删除整个值。请看我的代码，在代码列表中每秒保存每个传感器id的传入数据，为了减少内存存储，我必须在列表中只保留25秒的数据，有什么方法可以做到这一点吗？TTL清除整个列表。
public class ContinousDataProcessor
extends KeyedProcessFunction<String,SensorData,Tuple2<String,Integer>> {
private transient MapState<String, List&
浏览 3提问于2020-03-18得票数 0
1回答
在Flink中可以计算键控窗口的聚集输出吗？
我们有一个Datastream，我们调用byKey()指定一个由字符和数字组成的字段(例如，A01，A02.A10，B01，B02，B10等)，就像棋盘上的方格。在byKey()之后，我们调用window(TumblingEventTimeWindow.of(Time.days(7))，因此我们创建了一个每周窗口。在此之后，我们调用reduce()，从而得到SingleOutputStreamOperator<Result>。
现在，我们希望根据每个SingleOutputStreamOperator<Result>对象的一
浏览 5提问于2021-06-23得票数 0
回答已采纳
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
回答已采纳
在我们的应用程序中，flink检查点的大小正在增加，并且永远不会以rocks db作为状态后端。(AWS KDA)
Kafka-->do some magic -->ES(Sink)-->Writes to kafka
这里我们使用的键是UUID，并且不会重复。如何配置以确保检查点大小不会增加，或者对db进行微调以删除少于1天的任何较旧的密钥。
浏览 3提问于2021-03-01得票数 0
我在Flink上使用PrometheusReporter，并且有一些计数器我想用键控流的密钥导出。
我没有open上的密钥，所以我尝试使用processElement，如下所示：
// job
myStream.keyBy(Foo::getKey)
.process(new FooOperator())
.name("foo");
// operator
public class FooOperator extends KeyedProcessFunction<String, Foo, Bar> {
private trans
浏览 1提问于2021-03-11得票数 0
Flink Python文档中有pyflink.datastream.MapFunction。同时不存在RichMapFunction。谁能告诉我原因吗？
浏览 16提问于2022-05-05得票数 0
回答已采纳
我们正在构建一个flink用例，其中我们正在使用kafka主题，并根据平均、最大、最小阈值执行聚合和生成警报。当滚动事件时间Windows中有0个事件时，我们还需要为每个键发出通知。我们很难想出解决办法来做同样的事情。我们考虑的方案如下，请告诉我们，如果有其他的想法，我们还没有研究。
Queryable :在每个进程窗口函数中保存密钥。从外部应用程序查询状态，并在20分钟时间间隔过期后丢失密钥时发出警报。我们看到Queryable状态特性在未来将被废弃。我们不想走这条路，当我们已经知道有一个EOL为它。
使用处理时间Windows :如果下游应用程序将实时发送事件，则使用处理时间而不是
浏览 6提问于2022-05-10得票数 0
下面提到的Apache测试代码在我想使用KeyedOneInputStreamOperatorTestHarnessK、IN、Int或Long作为key (K)参数时不会在Scala中编译。可以使用字符串、整数或java.lang.Long类型的泛型参数编译相同的代码：
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.KeyedProces
浏览 7提问于2021-12-31得票数 2
我最近从Flink 1.2升级到Flink 1.3，我正在尝试更新我的ProcessFunction以使用1.3。我希望创建一个扩展ProcessFunction类的函数，但它抛出了一个编译错误，说明我没有覆盖processElement，onTimer是我拥有的代码：
class TimeoutStateFunction extends ProcessFunction[ObjectNode, (String, Long)] {
lazy val state : ListState[CountWithTimestamp] = getRuntimeContext
.getLi
浏览 18提问于2017-08-15得票数 0
回答已采纳
我在flink中有一个应用程序，它可以消除多个流的重复。它对一个字符串字段执行键操作，并使用值状态对其进行还原。
在RichFilterFunction中使用值状态。
public class DedupeWithState extends RichFilterFunction<Tuple2<String, Message>> {
private ValueState<Boolean> seen;
private final ValueStateDescriptor<Boolean> desc;
public Dedup
浏览 5提问于2022-05-16得票数 0
目前，Flink可以直接将数据写入hive的ORC格式的hdfs文件，但需要每小时向HIVE表插入分区。有没有办法每小时触发一次函数？
浏览 57提问于2020-07-18得票数 1
回答已采纳
我使用intellij、maven 3和flink 1.15.1编写有状态流作业。我正在尝试为我的自定义KeyedProcessFunction编写单元测试，并尝试遵循文档以及添加提到的的依赖项。我对使用KeyedOneInputStreamOperatorTestHarness类很感兴趣，但是在我的任何依赖项中都找不到这个类(下面发布的修改后的pom )。我唯一能找到这门课的方法就是
<dependency>
<groupId>org.apache.flink</groupId>
<ar
浏览 8提问于2022-08-01得票数 1
回答已采纳
2回答
异步闪烁计时器
、、
我有一个关于Flink和它的计时器服务的问题。
我有一个使用计时器的keyBy流，当计时器被调用时，我需要发送一个http请求，这可能需要一段时间来响应。我的问题是，我应该进行异步http调用吗？或者flink已经将计时器调用作为一个新线程，每个键都是异步的？
提前感谢
浏览 21提问于2020-02-17得票数 1
我需要编写一个批处理的flink作业，我更喜欢使用DataStream api。
我希望将作业的输出排序为每个输出文件(HDFS文件)。这样下游才能真正地利用这些来加速索引。
在中有的替代品吗？
我读过说，对于KeyedStream中的数据是按键二进制表示进行排序的，我可以使用这个特性吗？例如，创建一个由我想要排序的字段组成的字符串键？
浏览 6提问于2022-03-28得票数 0
这个例子一开始非常有用，它说明了keyedProcessFunction是如何在Flink工作的它来自Fraud检测器v2: State + Time部件，对于实际的应用需求部分，在这里设置一个定时器是合理的。
override def onTimer(
timestamp: Long,
ctx: KeyedProcessFunction[Long, Transaction, Alert]#OnTimerContext,
out: Collector[Alert]): Unit = {
// remove flag after 1 minute
timerSta
浏览 3提问于2020-05-26得票数 0
我有一个数据流，我在其中添加了一个RichFilter来过滤掉不合适的数据，同时使用flink状态。我想用一个键控处理函数来处理这个过滤出来的数据，因为我想在这个处理函数中使用flink valueState。但是因为filter的输出不是键控流，所以我不能用键控进程函数来链接它，除非我再次通过相同的字段键控它。 现在，这是有效的，但是有没有一种更简单的方法来做这件事，而不是键入两次？ DataStream stream = some stream data
stream.keyBy(myKeySelector).filter(RichFilterFunction).keyBy(myKeyS
浏览 22提问于2020-07-28得票数 0
1回答
某些事件的Apache延迟处理
、、、、
我有一个要求推迟处理其中的一些事件。
例如：我有三件事(刊登在卡夫卡上)：
A (id: 1，retryAt: now)
B (id: 2，retryAt: 10分钟后)
C (id: 3，retryAt: now)
我需要立即处理记录A和C，而记录B则需要在10分钟后处理。这在Apache中是可行的吗？
到目前为止，无论我研究了什么，似乎“触发器”可能有助于在Flink实现它，但还没有能够正确地实现它。
我也看过卡夫卡文档，但在那里似乎不可行。
浏览 7提问于2021-07-29得票数 2
回答已采纳
2回答
我有一个要求推迟处理其中的一些事件。
例如：我有三个活动(消费者卡夫卡)：
A (id: 1, retryAt: now)
B (id: 2, retryAt: 15 minutes later)
C (id: 3, retryAt: now)
我需要立即处理记录A和C，而记录B则需要在15分钟后处理。这是Flink能做的吗？
我尝试过用计时器使用KeyedProcessFunction来实现任务，但不幸的是不能工作。处理的事件不会发送到目标(接收器kafka生产者)。
守则如下：
你能帮帮我吗?
浏览 2提问于2022-04-07得票数 0
我们已经开发了一个Flink应用程序，它接收具有字段的消息，例如Flink应用程序中的id, value1, value2, ...，它们由id键控，并分配不同大小的滑动窗口和10秒的slide。我们使用AggregateFunction为每个id计算一些统计信息，结果数据流被汇聚到Redis。代码如下所示： DataStream<Tuple2<String, String>> statistics = messages.keyBy(0)
.timeWindow(Time.seconds(300), Time.seconds(10))
.aggrega
浏览 12提问于2021-01-12得票数 1
回答已采纳
1回答
我正在运行一个查询来连接一个流和一个表，如下所示。它正在耗尽堆空间。即使它在flink集群中有足够的堆空间(60 it * 3) 此查询是否需要驱逐策略？ SELECT sourceKafka.* FROM sourceKafka INNER JOIN DefaulterTable ON sourceKafka.CC=DefaulterTable.CC;
浏览 6提问于2019-09-26得票数 0
尽管Flink有一些内置工具来处理延迟数据，比如允许延迟，但我还是希望自己来处理延迟数据。例如，我希望监视延迟事件，或者将它们保存到数据库中。
我怎么能这么做？
浏览 0提问于2019-02-20得票数 1
回答已采纳
我想为KeyedProcessFunction中的所有状态保留一个简单的值，如下所示； class StateC() extends KeyedProcessFunction[Long, A, B] {
var timestamp: Timestamp = _
override def open(parameters: Configuration): Unit = {
timestamp = assignCurrentHour()
}
override def processElement(item: A, ... ): Unit = {
val curre
浏览 41提问于2021-09-12得票数 0
回答已采纳
我正在Flink数据流API版本1.10.1中运行TPC-H查询。其中一个UDF读取整个表LineItem并将其存储在内存中(使用默认的MemoryStateBackend)。首先，我存储了List<LineItem>对象的列表。但是我的内存快用完了，最好只存储我将使用的字段。所以我存储了List<Tuple2<Integer, Double>>。我还增加了发送方和接收方heartbeat.timeout: 100000请求和接收心跳的超时。当我使用大约500MB的数据源表文件时，我可以执行我的查询。然而，原始大小是725MB，在这个大小下，我遇到了一些指
浏览 12提问于2020-07-07得票数 0
回答已采纳
我使用Flink 1.7.2编写了一个简单的字数统计应用程序，同时使用Kafka 2.2作为消费者和生产者。我对Kafka生产者使用只有一次的语义，使用KeyedProcessFunction进行有状态处理，使用MapState保持我的状态，使用增量检查点的RocksDB作为我的状态后端。
当我在IntelliJ上运行这个应用程序时，它工作得很好，但是当我把它提交到我本地的Flink集群时，我收到了AsynchronousException异常，Flink应用程序每隔0-20秒就会重试一次。以前有没有人遇到过这个问题？从配置的角度看，我是否遗漏了什么？
下面是我的代码：
class Keyed
浏览 37提问于2019-07-05得票数 0
回答已采纳
Apache Beam最近通过StateSpec和@StateId注释引入了，并在Apache Flink和Google Cloud Dataflow中提供了部分支持。
我找不到任何与GlobalWindow一起使用时会发生什么的文档。具体地说，有没有一种“状态垃圾收集”机制，可以根据一些配置摆脱一段时间内看不到的键的状态，同时仍然为出现频率足够高的键保持一个完整的状态？
或者，在这种情况下使用的状态数量是否会发生变化，无法回收与一段时间内未见的键相对应的状态？
我还对Apache Flink或Google Cloud Dataflow中是否支持潜在的解决方案感兴趣。
Flink和direct
浏览 18提问于2017-03-14得票数 3
回答已采纳
1回答
在我的例子中，我只使用Flink的可查询状态。特别是，我不关心检查点。 在发生事件时，我仅在最多X分钟后查询可查询状态。理想情况下，我会删除“旧”状态以节省空间。 这就是为什么我想知道:我可以发信号通知Flink的状态在一段时间后自动清除吗？通过配置？通过特定的事件信号？多么?
浏览 20提问于2019-01-23得票数 1
当我想要更新值状态(queueState.update(Queue))时，捕捉这个异常： org.apache.flink.util.FlinkRuntimeException: Error while adding data to RocksDB
at org.apache.flink.contrib.streaming.state.RocksDBValueState.update(RocksDBValueState.java:108)
at xxx.xxx.xxx.CleanTimedOutPartialMatches.processElement(CleanTimedO
浏览 142提问于2019-10-12得票数 0
1回答
我试图开发下面的代码，但它不起作用。我想使用apache来延迟时间(在时间戳字段中指定)与当前日期不同的事件。
示例
当前日期: 2022-05-06 10:30
事件1[{ "user1"："1“、"user2"："2”、“时间戳”："2022-05-06 10:30“}-> OK (接收器)
事件2[{ "user1"："1"，"user2"："2"，“时间戳”："2022-05-06 13:30“}->延迟(将于13:30发送)
FlinkLikeP
浏览 7提问于2022-05-06得票数 0
我正在尝试一些网络监控工作的flink。我的目标是计算每个src_ip的不同src_ip。
下面的代码可以工作，但是性能确实很差。似乎每个滑动窗口重新计算所有事件，但这不应该是必要的。
例如，我们有活动准时第二次1- 600.Flink可以得到每秒钟的累加器，所以我们每秒钟有600个累加器。当第一个滑动窗口过期时，flink只合并1到300的累加器，并销毁第二个1的累加器。此窗口还可以在最后一秒钟前将1-299进行预合并。当第二个滑动窗口过期时，flink只合并2-301的累加器，破坏第二个的累加器。
这种方法比将事件分配给多个窗口和计算每个窗口的聚合效率要高得多。
支持flink吗？我能用f
浏览 1提问于2018-08-23得票数 3
当我提交新的flink作业时，它会抛出
Caused by: java.io.NotSerializableException: redis.clients.jedis.JedisCluster
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSer
浏览 18提问于2019-05-09得票数 0
回答已采纳
我有一个Apache usecase，它的工作原理如下：
我有数据事件通过第一流进入。每个事件的一部分是一个外键，我期望从第二个流中获得数据。例如:我得到的数据主要城市在第一流，其中有一个城市代码，我需要的平均温度随着时间的推移，这个城市代码流经第二流。不可能为所有可能的城市提供温度流，我们必须要求我们需要数据的城市。
所以我们需要一些方法来“通知”第二个流源，当我们在第一个流中第一次遇到它时，我们需要这个城市的数据“推送”。
如果可以从第一个流执行此通知，这将是很容易的。问题是，第二个流是通过websocket到达我们的，它的一部分是一个控制通道，我们必须通过它提出请求，所以请求必须从第二个
浏览 11提问于2022-01-24得票数 0
我正在扩展Apache的KeyedProcessFunction以定义工作流。我的工作流程由大约10-15个处理器组成。所有其他处理器collector.collect(T)在1秒内完成。而在最坏的情况下，一个特定的ProcessFuntion需要超过15秒。该进程函数发出与其他processFunction相同类型的有效负载。有效负载的大小也非常类似于其他处理器。在每个KeyBy之后，我还依赖于keyedProcessFunction ()。KeyBy()对所有流程函数都有相同的定义，并且在整个workFlow中依赖于相同的属性。
我如何调试/解决collector.collect花费这么多
浏览 2提问于2020-06-09得票数 1
回答已采纳
1回答
我有flink作业，我需要在1小时窗口内删除收到的记录的重复数据。一旦删除重复数据，我需要收集所有这些删除重复数据的文档，并进行一些聚合，如计数，然后生成目标主题。 附件是我的flink工作的图片： ? 现在，由于我只需要收集那些经过重复数据删除的文档，因此我可能不一定需要等待1小时。为了收集这些文档，我如何避免这种1小时的窗口，但是一旦收集到这些文档，就继续进行进一步的聚合。 由于这个原因，资源占用了内存，检查点大小也在增加，这是我想要避免的。 水印策略： WatermarkStrategy<PlImaGuidInterimMessageVO> watermarkStrat
浏览 19提问于2021-03-24得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
