我在连接的流上使用KeyedCoProcessFunction，两个流都以id为键，如果键不存在，我也使用MapState并设置一个类型为list的值，我也在processElement2中检查键的存在，所以理想情况下没有NPE的机会，但仍然可以得到它。lookupid"))) extend
浏览 51提问于2021-11-18得票数 0
在引用Flink代码示例时，我观察到当使用connect操作符连接两个流时，coProcessFunction和keyedCoProcessFunction在操作键控流(覆盖processElement1、processElement2和onTimer)时几乎可以互换和相同地扩展，所以当操作键控流时，与keyedCoProcessFunction相比，扩展CoProcessFunction来实现键控连接流的业务逻辑有什么不同
浏览 127提问于2020-09-02得票数 1
回答已采纳
我在使用Flink连接两个流的上下文中查看，并希望了解这两个流之间的区别以及影响Flink处理它们的方式。作为一个相关的问题，我还想了解CoProcessFunction与KeyedCoProcessFunction有何不同。
浏览 2提问于2021-02-17得票数 1
回答已采纳
我有两个流，一个数据流只包含一个设置ValueState passing=true/false的标志，另一个控制流添加了一个用户将被通知给MapState。当passing从false更改为true时，会向MapState中尚未收到通知的用户发出通知。
下面是处理此逻辑的KeyedCoProcessFunction。class TestKeyedCoProcessFunction extends KeyedCoProcessFunctio
浏览 1提问于2021-02-07得票数 1
回答已采纳
我使用KeyedCoProcessFunction函数来使用来自另一个流的数据丰富主数据。代码： packet: PacketData,
ctx: KeyedCoPr
浏览 2提问于2022-03-28得票数 1
回答已采纳
在用MiniClusterWithClientResource测试flink作业时，是否有办法控制触发 timer的处理时间？我能够测试KeyedCoProcessFunction的两种方法，即processElement().触发计时器回调，例如onTimer().在单元中，使用testharness测试并控制处理时间，即：//通过直接提高操作者的处理时间testHarness.setProcessingTime(300000)来触发处理时间定时器
因此。但是，我现
浏览 2提问于2020-08-25得票数 0
1回答
我正在阅读Kafka的多个主题，然后进行状态操作，然后再次保存到Kafka。taggedStream.getSideOutput(tag)
taggedStream.addSink(targetTopic) 在此流程中，延迟流的另一端： delayQueue(written by flink) => consumerAPP(check retrycounts and timestamps) => anotherQueue(这将再次
浏览 35提问于2021-04-24得票数 1
我有一个Flink KeyedCoProcessFunction，它在一个更大的Flink流作业中注册处理时间计时器，并且我正在尝试使用Flink 为整个作业创建单元测试。但是我不能让KeyedCoProcessFunction中的onTimer()回调触发。
有没有人把这个弄好了？它需要特殊的配置吗？切换到事件时间可以很好地工作，所以我想知道这是不是对Flink MiniCluster不起作用，或者我的</
浏览 0提问于2020-08-26得票数 3
对于1:1连接，我使用KeyedCoProcessFunction，我有两个流，查找流(每秒100条记录)和点击流(每秒10000条记录)。在processElement2方法中，我在MapState<Long,Row>中寻找关键字，如果找到的话，用它来丰富点击流数据，否则将此记录设置为端输出，然后将端输出设置为kafka。对于kakfa中的dlq主题，我连续看到每秒产生1-2条记录，在将其推送到端输出之前，我如何才能在processElement2方法中等待几毫秒的查找id
浏览 132提问于2021-11-11得票数 0
回答已采纳
我想在第二流的帮助下充实我的第一流，像流动的记录一样，像查找一样加入第二流，我想把它永远保存在记忆中，就像一张桌子。我可以使用的任何代码示例或flink API都适用于这个用例。
浏览 5提问于2022-05-30得票数 0
1回答
Flink是否保证流的执行顺序？我可以将parallelism设置为1来绕过这个问题，但是我想知道我看到的是否是预期的？return new JSONObject(e.f1).getString("someOtherKey");
source1.connec
浏览 11提问于2021-09-06得票数 0
我正在使用Flink来处理我的流数据。 我有两个数据源:A和B。dataA = env.addSource(sourceA);DataStream<String> dataB = env.addSource(sourceB); 我使用map处理来自A和B的数据我尝试做的是将它们合并为Aaaa, Bbbb, Cccc...以生成一个新的DataStream<String>对象。 如何做到这一点？
浏览 49提问于2020-07-23得票数 0
回答已采纳
今天，我想谈谈一个关于Flink的概念性话题，而不是一个技术性的话题。
在我们的例子中，我们确实有两个卡夫卡主题A和B，它们需要结合在一起。连接应该始终包括主题A中的所有元素，以及主题B中的所有新元素。实现这一目标有两种可能性:始终创建一个新的使用者并从一开始就开始使用主题A，或者一旦使用完就将主题A中的所有元素保持在一个状态内。主题A中的元素最终会丢失，如果窗口
浏览 8提问于2020-03-19得票数 0
回答已采纳
} } }在这里，在flatmap1方法中，我想更新一个数据库我可以在flatmap1中执行该操作吗?我之所以这样问，是因为查询db然后更新DB需要等待一些时间。
浏览 1提问于2021-01-22得票数 0
我正在编写一个Flink应用程序，它使用kafka主题中的时间序列数据。时间序列数据包含指标名称、标记键值对、时间戳和值等组件。我已经创建了一个滚动窗口，用于根据指标键(指标名称、键值对和时间戳的组合)聚合数据。我正在考虑使用flink的"allowedLateness“功能将延迟指标发送到不同的流。我计划在主"Aggregate the data“操作符中添加一个"MapState”，它将键作为度量关键字，将值作为到达主窗口的</
浏览 17提问于2021-08-09得票数 0
回答已采纳
1回答
我试图想出一个解决方案，其中包括在连接操作之后应用一些逻辑，从多个EventB中从streamB中选择一个事件。它类似于一个约简函数，但它只返回一个元素，而不是增量地执行它。因此，最终结果将是一个(EventA，EventB)对，而不是一个1 EventA和多个EventB的交叉乘积。假设像上面这样的连接操作，它用4 EventA生成了1个EventB，成功地加入并收集在MyJoinFunction中。现在我要做的是，立即访问这些值，并执行一些逻辑来正确地将EventA与匹配，这正是一个 EventB。因
浏览 2提问于2021-03-25得票数 0
回答已采纳
我的数据来自两个不同的Kafka主题，由不同的代理提供服务，每个主题具有不同数量的分区。一个流有关于正在提供的广告的事件，另一个流有点击：ad_clicks: ad_id, ip, cTime
流程函数的文档中有一节介绍了如何使用CoProcessFunction或KeyedCoProcessFunction实现，但我不确定如何设置。我还想知道Flink的是否
浏览 2提问于2021-07-15得票数 2
由于Model类(来自第三方库)不能由Flink自动序列化，所以我使用两个变量，如下所示：
class MyUDF extends KeyedCoProcessFunction[String, ModelDefmodelsBytes是真正的(键控)状态，它包含相同的模型，但作为一个字节块，这样检查点才能正常工作。总体解决方案很简单(只需要在恢复/保存模型时在模型上调用fromBytes/toBytes )，但我不知道这是否是一种常见的最佳实践。对于本质上相同的事情，有
浏览 3提问于2019-12-04得票数 1
回答已采纳
我在Apache中有两个DataStreams来自Flink训练 .window(TumblingEventTimeWindows.of(Time.milliseconds(2)))测试是可以的，但是在中我发现了与RichCoFlatMapFunction的连接，您能帮我理解一下：
哪种类型
浏览 3提问于2021-05-19得票数 0
云服务器
对象存储
ICP备案
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号