我使用Kafka Streams进行TimeWindow的一些聚合。我只对每个窗口的最终结果感兴趣，所以我使用了.suppress()特性，该特性为其状态创建了一个changelog主题。此changelog主题的保留策略配置被定义为"compact“，据我所知，它将至少保留过去每个键的最后一个事件。 我的应用程序中的问题是密钥经常改变。有没有办法告诉Kafka Streams从以前的窗口中删除密钥？在这个
浏览 33提问于2019-02-12得票数 0
回答已采纳
此时，我们可以假设我们有一个持久化的kafka主题和一个rocksDB本地存储库，用于A-KTable和B-KTable聚合。尽管如此，我的问题是A-KTable和B-KTable物化生命周期(即changelog主题和本地rocksdb存储)。假设A-Event主题和B-Event主题保留策略被设置为2周，是否有一种方法可以副作用卡夫卡内部KTable物化主题保留策略(changelog和rocksDB)与上游事件主题删除保留<em
浏览 1提问于2019-05-21得票数 0
回答已采纳
1回答
我们正在与KSQLDB一起执行POC，并有一些疑问：
我有一个名为USERPROFILE的卡夫卡主题，它有大约1亿张独特的记录和10天的保留政策。这个Kafka主题继续从其底层的RDBMS表接收实时插入/更新类型的事件。userprofile_stream (userid INT, firstname VARCHAR, lastname VARCHAR, countrycode VARCHAR, rating DOUBLE比如说，我们在5月1日收到了id为1001的记录，然后在5月11日，卡夫卡主题上的记录将不复
浏览 3提问于2021-05-27得票数 1
我使用kafka流约函数，它创建了一些状态存储、更改日志、kafka内部主题(比如app-KSTREAM-REDUCE-STATE-STORE-0000000002-changelog )。我希望设置保留字节，并将清理策略更改为delete，以防止存储空间被填满。TopicConfig.CLEANUP_POLICY_DELETE);但是，当生成一个新主题时，只会将保留
浏览 0提问于2018-09-02得票数 6
回答已采纳
所需的功能:对于给定的键key123，多个服务并行运行，并将其结果报告到单个位置，一旦为key123收集了所有结果，这些结果就会传递给新的下游使用者。新想法:使用Kafka Streams和KSQL来达到相同的目标。所有服务都将它们的输出写入结果主题，该主题形成了一个更改日志Kstream，我们可以在KSQL中查询已完成的条目。类似于：
CREATE STREAM competed_results FROM results_stream SELECT * WHERE (all results != NULL).
浏览 9提问于2019-07-12得票数 0
我们有一个Kafka流聚合拓扑。我们需要控制changeLog主题的大小，以降低Kafka存储成本。但是它是否也从changeLog主题中删除了该记录呢？更重要的是，它是否也减少了changeLog主题的大小，从而使Kafka存储成本得到控制？
浏览 3提问于2020-08-29得票数 2
1回答
但是，我遇到了一个关于状态存储和更改主题保留策略的问题。用户会话可能在时间上无限期地发生，因此，我必须保证状态将在此期间持久存在，并在节点和集群范围内发生故障时恢复。(KAFKA-4015是在0.10.1版本中修正的，加窗口的变更主题在应用额外的保留时间参数时不会无限增长)。
“对于有窗口的KTables有一个本地保留时间和一个变更日志保持时间。如果创建了新应用程序，则会使用与本地存储保留时间相同的保留时间创建changelog主题。看起来，<e
浏览 5提问于2020-03-04得票数 1
回答已采纳
此窗口策略保留来自流的信息，其类型数量与消息加入或未加入的事实无关。
在输入流吞吐量很高的情况下，由Kafka创建的用于执行连接的主题可能增长很快，占用了大量的磁盘空间。
浏览 2提问于2019-01-28得票数 2
回答已采纳
2回答
我读了很多关于kafka用作活动商店和CQRS的潜在候选者的文章。我想知道，由于kafka中的消息有一个有限的保留时间，那么消息从kafka保留消息的磁盘中删除后，事件将如何重播？从逻辑上讲，当这些消息从kafka外部(在读取kafka主题的消息之后)存储在数据库(sql/nosql)中时，从事件存储的角度来看，这比kafka更有意义。相反，如果我的理解是正确的，那么在CQRS中使用kafka的真正用例是什么，尽管k
浏览 3提问于2019-11-16得票数 0
或者，例如，我们可以将数据保留一个月，但只在过去一周加入流？
有没有什么好的例子来展示开窗口的KStream- to -kStream窗口连接？
浏览 1提问于2017-01-18得票数 9
回答已采纳
1回答
如我所想，KTable -带有compaction删除策略的简单kafka主题。此外，如果为KTable启用了日志记录，则还会有changelog，然后，删除策略为compaction,delete。在这两种情况下，我们都会得到一段时间内key的最后一个值(?)。本地存储用于聚合步骤、连接等。但在此之后还会创建具有压缩策略的新主题。例如：
KStream<K, V> sour
浏览 2提问于2018-09-25得票数 19
回答已采纳
在我们的事务处理系统中实现了kafka流处理。该解决方案实现如下：
Kafka producer将事件发布到kafka topic，流处理器处理输入事件并进行聚合操作。
浏览 0提问于2017-08-14得票数 1
定义了一个自定义存储，在自定义Transformer中使用(参考如下)。KStream<String, String> testStream = kstreamBuilder.stream(Serdes.String(), Serdes.String(), "test");[2018-05-14 23:38:09,391] ERROR stream-thread [StreamThread-1] Failed to create an active task 0_1:&#x
浏览 2提问于2018-05-15得票数 0
/bin/kafka-topics.sh --zookeeper zookeeper:2181 --alter --topic topic_1 --config retention.ms=3600000/bin/kafka-topics.sh --zookeeper zookeeper:2181 --describe --topics-with-overrides
Topic:__consumer_offsets
浏览 0提问于2018-11-15得票数 1
回答已采纳
我们正在使用spring cloud stream和Kafka 2.0.1，并利用InteractiveQueryService从商店获取数据。在聚合数据后，有4个存储将数据持久化到磁盘上。请注意，设置为两个月数据的保留和清理策略将被删除。我们不压缩数据。 磁盘上各个状态存储的大小在14 gb到20 gb之间。anomaly-timeline-twelve-hours-error-score-changelog-3.我确实将auto.offset.reset设置为“最早
浏览 18提问于2019-03-08得票数 4
我们通过集成使用Kafka流。我将复制因子配置为在所有内部Kafka流主题中使用，方法是设置它适用于大多数由Kafka流在内部使用的重新分区/变更主题。对于这些主题，我仍然可以看到复制因子设置为默认1，也不可能使用Materialized#withLoggingEnabled(Map
浏览 5提问于2020-10-14得票数 1
回答已采纳
我想将这些消息保存在状态存储中，但是Kafka流无法生成到内部changelog主题：2017-11-17 08:36:20,583 ERROR StreamT
浏览 1提问于2017-11-17得票数 3
回答已采纳
我有一个使用Spring Cloud Stream Kafka的应用程序。对于用户定义的主题，我可以通过下面提到的配置从指定的主题中删除记录。但此配置不适用于DLQ主题。例如，在下面的配置中，我在绑定器级别配置了保留时间。因此，在绑定级别下定义的我的生产者主题(学生主题)配置正确，当主题日志超过指定的保留字节(300000000)时，我可以检查记录是否被删除。但是绑定器级别的保留时间不适用于DLQ主题(person-topic-error-dlq)。除保留</e
浏览 51提问于2021-03-12得票数 1
回答已采纳
我有一个Kafka读数流，我会检查是否超过了某个阈值。我只想在第一次超过警报时传播警报。为此，我首先计算新状态，将新状态分组到KGroupedStream中。然后减少到一个KTable，我在这里检查状态是否发生了变化(保留一个布尔值)，并更改为changelog流，并对状态发生变化的记录进行过滤。我的理论是，这应该是可行的，但是并不是每个状态更改都会传播到changelog流，而只是偶尔更新一次changelog流(无法真正看到模式)。有人知道为什么会这样吗，或者更好的是我如何解决这个问题？简化示例：
K
浏览 28提问于2017-08-10得票数 1
回答已采纳
1回答
我正在尝试为log.retention.ms、log.retention.bytes、log.segment.bytes和log.segment.delete.delay.ms设置不同的值。首先，我将所有值设置为允许的最小值，但这似乎降低了性能，然后将它们设置为代理在完全满之前可以接受的最大值，但同样，当发生删除时，性能也会下降。是否有设置这些值以获得绝对最小延迟的最佳实践？
浏览 0提问于2017-02-18得票数 11
回答已采纳
云服务器
ICP备案
对象存储
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号