kafka Streams配置： threads = 1;ktableCommitInterval= 10000;
ktableMemory=72000000问题： 提交间隔设置为10秒。高速缓存大小被设置为72MB。数据以字节为单位。状态存储已启用缓存。文档指出，kafka streams将数据推送到下游的操作语义取决于缓存大小或提交间隔，无论首先发生什么。但根据实验，提交立即发生了两
浏览 30提问于2020-01-21得票数 0
TimeWindows.of(TimeUnit.MINUTES.toMillis(60)), "HourlyPageViewsAgg")我现在的问题有两个：
考虑到我有自己的Cassandra驱动程序，我如何编写聚合到Cassandra的5分钟中间结果？阅读可以看出，如果不使用低级别的API，它可能是无法避免的，因为这似
浏览 2提问于2017-09-07得票数 0
电流配置
应用程序运行在应用程序中的Flink 1.14.4上，数据流(由长链操作符产生)间隔连接到一个Kafka源。事件时戳每个分区严格上升。默认情况下， Kafka源(间隔连接的右侧)比其他数据流(kafka > source )每1分钟生成一次数据，因此它的分区最多可被视为空闲59秒。区间连接操作符的输出应该是(并且曾经)两个传入流的最小值(卡夫卡和DS)，但是对于KafkaSource，它在一定时间之后被设置为最大。因此，数据流中的所有记录都被删除，<em
浏览 12提问于2022-06-17得票数 0
回答已采纳
https://kafka.apache.org/21/javadoc/org/apache/kafka/streams/state/Stores.html#persistentWindowStore-java.lang.String-java.time.Duration-java.time.Duration-boolean- 挖掘streams内部代码似乎是用来设置一些序列号的？
浏览 26提问于2019-04-10得票数 5
我目前正在使用kafka streams来整理窗口中的相关事件。如果所有相关事件没有在一个窗口内到达，在Kafka streams中是否有一种方法可以让我们获得过期事件的句柄。下面是一些例子GroupID: g1事件到达时间: E1,10am；E2 10:01 are和E3 会话窗口的非活动持续时间为5分钟。结果:所有事件都整理成功。示例2：
不活动状态持续时间为5 m
浏览 1提问于2017-03-31得票数 3
1回答
我想知道是否有任何方法可以使用Kafka流、DSL或处理器API对窗口中的记录进行排序。然后，我们使用这个重新分区的流，并将消耗的记录存储在本地状态存储中，窗口为10分钟。特定用户的所有单击总是在同一个分区中，但顺序不受保护，因为最初的主题有10个分区。我理解Kafka流的窗口模型，当新记录出现时，时间会提前，但我需要这个窗口使用处理时间，而不是事件时间，然后当窗口过期时，我需要能够对缓冲事件进行排
浏览 2提问于2018-03-13得票数 7
1回答
我想在我的kafka-streams应用程序中消除重复，它使用状态存储，并使用这个非常好的例子：
我的第二个问题这个时间间隔是否相关，应该
浏览 4提问于2020-08-01得票数 1
4回答
同步来自多个数据源的数据
、、、、
我们正处于设计阶段，目前的系统设计如下：
为了将一组事件分类为异常，必须在同一时间窗口中发生事件。例如，有三个数据源将各自的事件推入Kafka主题，但由
浏览 0提问于2019-05-27得票数 8
回答已采纳
1回答
我正在窗口从流环境传入的对象，收集和打印它。使用卡夫卡压制，以避免中间结果。import com.savk.workout.kafka.streams.kafkastr
浏览 5提问于2019-11-05得票数 3
回答已采纳
key1", json, now.plusMillis(2000L).toEpochMilli()));
然后我试着比较结果
浏览 2提问于2020-05-29得票数 2
我在kafka流中有以下逻辑流： .foreach((k, v) -> print(k.window().start()));
传递给count方法的参数与上面显示的相同，因此，它应该将键聚合到(逻辑)小时窗口中为了验证这一点，我在foreach调用中打印窗口启动时间(或者至少这是我所期望的)。我的COMMIT_INTERVAL_MS_C
浏览 4提问于2017-10-15得票数 0
回答已采纳
2回答
我无法理解kafka流中的groupBy/groupById和窗口的概念。我的目标是在一段时间内(例如5秒)对流数据进行聚合。{"value":0,"time":1533875665509}{"value":8,"time":1533875669512}
时间以毫秒为单位在这里，我的时间戳在
浏览 0提问于2018-08-10得票数 7
我想实现的目的是按用户分组我从Kafka主题收到的一些消息，并对它们进行窗口，以便聚合我在(5分钟)窗口中收到的消息。然后，我想收集每个窗口中的所有聚合，以便立即处理它们，并将它们添加到我在5分钟间隔内收到的所有消息的报告中。streams.start();
浏览 0提问于2018-07-19得票数 8
回答已采纳
我使用的是翻滚窗口(5分钟)和AscendingTimestampExtractor，因为我的源码来自Kafka。但是窗口的运行时间总是超过5分钟。有没有人能建议一下？
浏览 13提问于2020-02-25得票数 0
我使用kafka流组件在30分钟的滑动窗口上构建聚合(和)，宽限期为2分钟。我正在按10,000个时间序列(组)进行处理。聚合使用禁用日志记录的持久状态存储。若要在聚合间隔结束时仅输出最终结果，我将使用被抑制的运算符。我的问题：
如何正确调整记录缓存的大小以确保正确计算所有
浏览 3提问于2021-04-28得票数 1
有一个Kafka流应用程序，它通过流连接执行窗口(使用原始事件时间，而不是挂钟时间)，例如1天。如果打开这个拓扑，并从一开始就重新处理数据(就像在lambda风格的体系结构中那样)，那么这个窗口会将旧数据保存在那里吗？使用Kafka Streams 2.5.0进行更新
浏览 6提问于2022-01-09得票数 0
(JsonDeserializer.java:34) at org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:131)
at org.apache.kafka.streams.state.internals.Name
浏览 2提问于2017-04-28得票数 0
回答已采纳
2回答
我计划建立一个MySQL to Kafka流程，最终目标是安排一个流程，根据更改后的数据重新计算mongoDB文档。
这可能涉及直接修补mongoDB文档，或者运行将重新创建整个文档的进程。有没有办法“揭穿”Kafka流？例如，对于Kafka消费者，有没有一个定义良好的模式可以用来实现我想要的逻辑？
浏览 8提问于2021-07-06得票数 0
我有一个卡夫卡流应用程序，目前只是加入两个KStreams与一个5分钟的窗口，并将加入结果的另一个topic。我原以为磁盘空间的利用率是稳定的。但是，情况似乎并非如此。如果我在kafka streams应用程序集群中的一个应用程序失败了
浏览 2提问于2019-08-27得票数 1
回答已采纳
1回答
在我的应用程序中，我想要一个div弹出窗口后，用户登录，该弹出窗口将显示每小时的登录时间。div弹出窗口将显示一个文本框并提交表单，如果用户在第二个div弹出窗口之前没有提交任何数据，则应自动提交表单，并将数据存储到数据库中。现在的问题是如何显示div弹出每小时，如果我使用倒置在javascript其唯一的客户端和刷新的间隔值将被更改，所以javascript将不会used.second选项是使用会
浏览 2提问于2015-01-06得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
