在KeyedStream.process之后assignTimestampsAndWatermarks不起作用
的原因可能是因为在KeyedStream.process操作之后，数据流的时间戳和水印信息已经被修改或丢失。这可能是由于在process操作中没有正确处理时间戳和水印，或者在process操作中对数据进行了重新分区或重新排序等操作，导致assignTimestampsAndWatermarks无法正确应用。
为了解决这个问题，可以尝试以下几个步骤：
- 确保在KeyedStream.process操作中正确处理时间戳和水印。在process函数中，可以使用Context对象来访问时间戳和水印信息，并根据需要进行处理。例如，可以使用Context.timestamp()获取事件的时间戳，并使用Context.timerService()来注册定时器或更新水印。
- 如果在process操作中对数据进行了重新分区或重新排序等操作，需要确保在重新分区或重新排序之后重新应用assignTimestampsAndWatermarks。可以使用KeyedStream.rebalance或KeyedStream.rescale等操作来重新分区，然后再调用assignTimestampsAndWatermarks方法重新应用时间戳和水印。
- 检查是否存在其他操作或转换在KeyedStream.process之后修改了时间戳和水印信息。例如，某些操作可能会清除或重置时间戳和水印信息，导致assignTimestampsAndWatermarks无效。确保在process操作之后没有对数据流进行这样的操作。
总之，要确保在KeyedStream.process操作之后assignTimestampsAndWatermarks能够正确起作用，需要正确处理时间戳和水印，并避免在process操作之后对数据流进行修改或丢失时间戳和水印信息的操作。
相关·内容
keyBy工作前的assignTimestampsAndWatermarks： DataStream<Trip> trips = featurizedUserTrips.timeWindowAll(Time.days(7),
Time.days(1)); 但不是在keyBy和process之后 DataS
浏览 98提问于2019-04-17得票数 1
我正在处理一个项目，并打开了Flink选项disableAutoGeneratedUID，以确保所有内容都具有适当的uid。但是，当我试图运行以下工作时，我遇到了一个错误：一年前，我在网上看到
浏览 1提问于2021-06-11得票数 0
回答已采纳
1回答
我正在尝试使用Apache流API连接两个流，但是没有加入任何东西，在阅读了文档之后，我不知道我做错了什么 .assignTimestampsA
浏览 1提问于2016-04-26得票数 1
回答已采纳
//to some reducer new BoundedOutOfOrdernessTimestampExtractor<Metric
浏览 17提问于2021-07-24得票数 0
回答已采纳
我正在尝试设置一个流，其中首先执行一个countWindow。countWindow发出的结果需要传递给不同的timeWindow。问题是timeWindow没有发出任何结果。 我给出了一个非常简单的代码来演示这个问题： val env = StreamExecutionEnvironment.getExecutionEnvironment .addSource(new RichSourceFunction[Int] {
override def
浏览 24提问于2019-09-02得票数 0
回答已采纳
AvroDeserializationSchema.forSpecific(MyPojo.class),);
myEvents.assignTimestampsAndWatermarkswithTimestampAssigner((evt, timestamp) -> evt.event_timestamp_field)); Q.1根据我正在阅读的内容，我的时间0-60的窗口将在90秒后计算，30-90在120然而，
浏览 33提问于2021-07-08得票数 0
回答已采纳
2回答
我想加入两个来自卡夫卡制片的流，但加入不起作用。我使用AssignerWithPeriodicWatermark来定义我的指定者，并尝试使用3分钟窗口连接这两个流。但我没有得到任何输出。我打印了这两个流，以确保它们的事件在时间上足够接近。String]("topic_id", new SimpleStringSchema(), properties)
val streamId = env.addSource(consumerId).assignTimestampsAndWatermarksString]("t
浏览 2提问于2020-04-04得票数 0
回答已采纳
3回答
我需要在等待一段时间后关闭此窗口，后续窗口可以在系统恢复后继续。我如何才能做到这一点？kafkaDataStream .filter(Objects::nonNull)
.assignTimestampsAndWatermarksSerializableTimestampAssigner<Entity>) (element, recordTimestamp) -> element.getTimestamp()))
浏览 0提问于2021-01-04得票数 0
水印将如预期的那样被保留为0：我还可以看到在水标记中生成的这些日志watermark: 1601475710619event watermark: 1601475710619有趣的是，当我在IntelliJ上本地运行相同的应用程序，并且在相同的主题上有空闲的kafka分区时，我也会从水标中
浏览 2提问于2020-10-01得票数 0
回答已采纳
下面是我的Flink工作流程：org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException:无法将元素转发到下一个运算符
在flatMapQ1:我想复制是因为下游操作符在作业重新启动后重复使用消息。我说得对吗？我在<
浏览 7提问于2019-12-27得票数 3
回答已采纳
1回答
我在Centos服务器上，当我试图运行即使在尝试添加Permission Denied之后，也会得到chmod +x script.sh错误。
浏览 0提问于2018-10-01得票数 -1
1回答
有人能解释一下为什么我的地图在模板中不起作用吗？ 我试了4个小时，但没有得到一个有用的解决方案。
浏览 20提问于2021-10-21得票数 0
我想知道，在kafka中，水印的时间戳将使用消费者的时间还是生产者的时间？FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), properties) .assignTimestampsAndWatermarks
浏览 78提问于2021-08-12得票数 0
回答已采纳
使用TumblingProcessingTimeWindow和apply()中的自定义连接逻辑在同一流上窗口。即使所有事件都是成对的，也不会立即触发此窗口。
浏览 3提问于2020-06-18得票数 0
回答已采纳
8回答
' ') { } sTmp[j] = '\0';}
在我输入"loops“之后，"s1”被自动分配了一个空行。
浏览 0提问于2011-05-07得票数 19
回答已采纳
在我的应用程序中，使用以下代码激活滚动视图。scroll.autoresizesSubviews=YES;但在使用UIImagePicker/Controler.ie之后从手机画廊中选择图像后，滚动功能不起作用。上面的代码是用viewWillAppear section.Can写的，有没有人能帮我，让我在UIImagePickerController之后也能滚动？
浏览 2提问于2013-05-13得票数 1
1回答
在命令执行之后，我执行popen()和fgets()函数来获取已经启动的vlc应用程序的PID。但是fgets()并没有给出PID。代码如下所示。它是从“root”运行的。printf("\n%d\n", atoi(arrcPID));else iRetVal = -1;}
注意:这段代码在solo但它在我的实际代码库中不起作用，因为我的代码库非常大，而且其他popen()函数也被多次调用。我不能理解这个问题。请帮
浏览 2提问于2014-10-14得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
