我正在使用Streams DSL和状态聚合(从一个主题读取数据，聚合数据并将数据写入另一个主题)。如何缩短写入状态存储的数据的保留期？现在，我的基础架构团队说，数据在状态存储中保留了5年，我必须减少这一数字。是否有特定的配置可以设置数据应该保留多长时间？
浏览 11提问于2019-05-23得票数 2
1回答
我目前正在研究一个跟踪用户与平台交互的用例，从而生成一个事件流，这些事件流存储在kafka中，随后将在Kafka Streams/KSQL中进行处理。(.)Kafka Streams DSL使用的默认实现是一个容错状态存储，使用1.内部创建和压缩的变更主题(用于容错)和2.一个(或多个) RocksDB实例(用于缓存键值查找)。如果创建了新应用程序，则会
浏览 5提问于2020-03-04得票数 1
回答已采纳
我的印象是，我的应用程序中的状态商店在每5分钟的窗口之后都会被修剪(因为我的应用程序只关心加入状态的5分钟事件窗口)。
什么时候卡夫卡流应用程序清理它的状态？如果我在kafka streams应用程序集群中的一个应用程序失败了，然后我启动另
浏览 2提问于2019-08-27得票数 1
回答已采纳
我使用kafka流约函数，它创建了一些状态存储、更改日志、kafka内部主题(比如app-KSTREAM-REDUCE-STATE-STORE-0000000002-changelog )。我希望设置保留字节，并将清理策略更改为delete，以防止存储空间被填满。因此，我在kafka代码中设置了以下内容：props.put(StreamsCo
浏览 0提问于2018-09-02得票数 6
回答已采纳
1回答
我有一个kafka streams的用例，其中我需要对过去的数据执行聚合操作，这些数据甚至可能在几个月前被消耗。我想知道这是否意味着我需要关注内部主题的默认保留期，例如XXX-REDUCE-STATE-STORE-changelog、XXX-AGGREGATE-STATE-STORE-repartition和以某种方式显式更改/设置？如果我在broker级别设置默认保留期，我新创建
浏览 21提问于2021-02-01得票数 0
在默认情况下，Kafka Streams州立商店是“紧凑的”。是否可以在状态存储中设置带有保留策略的“紧凑、删除”？
浏览 0提问于2018-05-31得票数 5
回答已采纳
1回答
我想加入两个流，我已将连接窗口设置为25小时，因为要加入的记录最多可相隔24小时。JOIN_WINDOW),这将引发以下异常：
org.apache.kafka.streams.errors.TopologyException:无效拓扑:连接窗口KSTREAM-JOINTHIS-0000000016-存储的保留期不能小于其
浏览 0提问于2018-12-31得票数 3
回答已采纳
由于某些原因，在保留策略到期后，我的旧状态存储没有进行清理。我在本地测试它，所以我只在大约5分钟内发送一条测试消息。我将保留时间设置得很低，只是为了测试。retentionPeriod = 120，retentionWindowSize = 15，我假设保留重复项应该是false。什么时候才是真的呢？Duration.of(retentionWindowSize, ChronoUnit.SECONDS),当
浏览 0提问于2019-04-03得票数 0
我将KafkaStreams 0.10.2.1与窗口化的RocksDB状态存储一起使用，并且在状态存储初始化过程中看到了非常奇怪的行为。在每个任务的状态存储文件夹中，KafkaStreams正在创建和删除包含RocksDB文件的文件夹30分钟。
如果状态存储名为XXX，则在名为XXX的文件夹中创建文件夹。State Folder
浏览 5提问于2017-10-23得票数 2
回答已采纳
当您运行Kafka streams应用程序重置工具以将应用程序重置为特定的时间戳(例如T-n)时，状态存储会发生什么？文档内容为：“内部主题:删除内部主题(这会自动删除任何已提交的偏移量)”(内部主题在执行状态存储的changelog主题时由Kafka Streams应用程序在内部使用)
这是否意味着我丢失了状态存储/RocksDB的<
浏览 13提问于2020-02-25得票数 1
回答已采纳
在启动卡夫卡消费的同时，获得低于例外的地位。org.apache.kafka.clients.consumer.OffsetOutOfRangeException:偏移超出分区{test-0=29898318}的范围，没有为分区配置重置策略。Kafka版本: 9.0.0 Java 7
浏览 13提问于2016-05-19得票数 19
是否可以在单个事务中生成Kafka主题并写入状态存储？而不是作为主题消费的一部分启动事务。我也知道Confluent提供的解决方案。但是，只要“从主题到商店”的同步时间小于阻塞时间，这就
浏览 3提问于2019-03-01得票数 0
2回答
我有默认7天的最新流媒体数据存储在Kafka中：在部署新版本的Streams应用程序时，需要大量时间处理旧数据，然后才能实际使用它。除了缩短留用期外，还有其他办法可以让它更快吗？
我想到的是，在所有数据被处理之前，状态存储不应该持久化到磁盘。
浏览 7提问于2022-02-21得票数 0
我面临以下问题:改变一些与kafka相关的属性，并重新启动集群。
In kafka Consumer, there were 5 consumer jobs are running .如果我们做了一些重要的属性更改，并且在重新启动集群时，一些/所有现有的使用者作业都无法启动。
浏览 1提问于2017-06-22得票数 2
回答已采纳
我需要监视我的应用程序的容错状态存储的恢复过程。在汇流中有一个例子。为了观察所有状态存储的恢复，您可以为应用程序提供一个org.apache.kafka.streams.processor.StateRestoreListener接口的实例。通过调用org.apache.kafka.streams.processor.StateRestoreListener方法来
浏览 0提问于2020-02-19得票数 1
回答已采纳
我有一个函数，它接受Kafka streams的一个实例，获取状态存储，解析它并进行一些计算。void func1(KafkaStreams streams) StoreQueryParameters<ReadOnlyKeyValueStore<String, Long>> storeQueryParams.
// using this iterator, I will read each record in state sto
浏览 24提问于2020-10-09得票数 0
我目前正在使用kafka streams来整理窗口中的相关事件。如果所有相关事件没有在一个窗口内到达，在Kafka streams中是否有一种方法可以让我们获得过期事件的句柄。感谢您的回复。示例1：不活动状态持续时间为5 mins.Result:的</e
浏览 1提问于2017-03-31得票数 3
我们在项目中使用Kafka流状态存储，我们希望存储超过1MB的数据，但是我们得到了下面的例外：
topic.max.request.size=50000000
然后应用程序可以正常工作，但是当创建了状态存储内部主题时，但是当Kafka重新启动并且状态存储<
浏览 3提问于2020-06-04得票数 1
我使用Kafka Streams进行TimeWindow的一些聚合。我只对每个窗口的最终结果感兴趣，所以我使用了.suppress()特性，该特性为其状态创建了一个changelog主题。此changelog主题的保留策略配置被定义为"compact“，据我所知，它将至少保留过去每个键的最后一个事件。 我的应用程序中的问题是密钥经常改变。有没有办法告诉Kafka<
浏览 33提问于2019-02-12得票数 0
回答已采纳
2回答
嗨，我有一个类似于下图所示的架构。有没有一种方法可以让我以一种简单的方式处理这种情况，比如服务总线主题。
浏览 1提问于2018-01-03得票数 5
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
