我将一些记录放入PersistentKeyValueStore：KeyValueIterator<String, Long> iter = myStore.all(); System.ou
浏览 1提问于2018-01-19得票数 1
回答已采纳
在我的用例中，当我从其他微服务接收客户事件时，我需要存储在customer-order customer 物化视图中，当我接收到 order 事件时，我需要加入客户并订购，然后存储在物化视图中。StoreBuilder customerStateStore = Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("customerorg.apache.kafka.streams.state.internals.QueryableStoreProvider.getSt
浏览 0提问于2018-04-24得票数 0
回答已采纳
]]]("store-name") ).to(outTopicName) 在“聚合器”函数中，无论tombstone事件如何，这些主题都会不断增长。我找不到一种方法来清理它们。 我们的要求很简单： 只要满足键的条件，该键的聚合changelog就没有用处。我们希望彻底永久地清除所有事件中的该密钥 请建议如何清理基于key的kstream内部主题？非常感谢。
浏览 30提问于2020-10-08得票数 0
1回答
问题：在一个JVM中拥有一个org.apache.kafka.streams.KafkaStreams类的多个实例意味着什么(EG:内存、CPU使用、并发问题)？后台：我试图提供一个散列机制，这样如果流操作抛出异常，它就不会将整个KafkaStreams实例转换为ERROR状态。我将应用程序划分为不同的KafkaStreams实例，每个实例负责不同的任务(日志记录、外部web调用、db调用等)。我还没有找到有关以下方面的文档：(1)从KafkaStreams状态恢复ER
浏览 2提问于2018-03-15得票数 1
回答已采纳
我想创建一个基于Kafka流的应用程序，该应用程序处理一个主题并以X(即50)大小的批次接收消息，但如果流的流量较低，则在Y秒内(即5)给出该流的所有内容。这是为了使一些I/O绑定处理更有效。
浏览 1提问于2018-12-05得票数 2
回答已采纳
我不确定我是否理解kafka流指标是如何导出的。actuator可以获得这些JMX指标吗？有没有办法获得这些指标并以普罗米修斯格式公开它们？
浏览 46提问于2019-09-25得票数 7
为此，我使用会话窗口，因为这符合要求，在这里，我必须将具有唯一事务id的事件聚合到列表中。每个事务都可以有由不同应用程序生成的多个事件，并将它们推送给kafka。kafkaStreams = new KafkaStreams(builder.build(), streamsConfig);留用期在这里是如何运作的此时，上述记录应从状态存储中过期，作为结束时间<流时间保持期(30秒)。
3)接下来，我摄入了与
浏览 0提问于2018-04-17得票数 2
回答已采纳
2回答
如何检查KeyValueStore是否已完全填充？如果我通过InteractiveQuery查找密钥，我需要知道密钥是否不存在，因为StateStore还没有准备好，或者密钥是否确实不存在。
浏览 1提问于2020-03-11得票数 4
回答已采纳
1回答
这个内存中的键值存储是在grpc调用期间调用的.在再平衡期间(如果消费者消失了)，数据将被写入到其他使用者的本地存储中。
如果消费者运行了大约2周，那么这些服务似乎没有内存了。我们可以删除不再需要的分区的数据吗？还是有一种方法可以在恢复使用者后删除存储的数据？
浏览 3提问于2019-12-10得票数 1
回答已采纳
我正在一个服务中运行多种kafka流实例。我希望动态地调整num.stream.threads配置，以便在程序运行时控制每个实例的优先级。
我没有在KafkaStream类上找到相关的方法。
浏览 2提问于2019-01-04得票数 1
回答已采纳
如何从KafkaStream获取主题名称和分区id。partition id= %s, offset = %s%n",record.key(), record.value(), record.partition(), record.offset());}
不知道如何在KafkaStreams中获取记录引用。
浏览 2提问于2020-06-20得票数 4
回答已采纳
在Scala中，我使用import org.apache.kafka.streams.KafkaStreams，能够读取输入流，并且可以轻松地进行一些计算并发送到输出流。是否有办法通过branch或filter从输入流中获取结果记录并将两个输出流发送给？
浏览 0提问于2020-02-12得票数 0
回答已采纳
我希望有人能帮我解决卡夫卡中关于GlobalKTables的一个问题。SELECT * FROM GlobalKTable where table_key.contains("055”) OR table_value.contains(“055”)
我已经翻阅了官方文档，但没有找到如何做到这一点的例子。我怀疑<e
浏览 0提问于2020-04-23得票数 1
StoreBuilder countStoreBuilder = Stores.persistentKeyValueStore("Counts我只是在接收记录，并将其中一些记录与存储中的先前项目进行比较，并将收到的一些记录存储在国家商店中。开发人员指南提到，您可以通过在上面的构建器上调用.withCachingEnabled()来启用处理器API中的记录缓存。
缓存“用作
浏览 0提问于2019-05-29得票数 5
1回答
试图通过KTable从“连接-信任”主题中获取记录 KTable<String, Object> ktable = builder.table("connect-configs"); streams.cle
浏览 0提问于2018-04-18得票数 3
回答已采纳
当我通过@Service (服务从状态存储获取聚合数据)获取数据时，我在单元测试中遇到了这个问题。org.apache.kafka.streams.state.internals.QueryableStoreProvider.getStore(QueryableStoreProvider.java:60) at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1043) at org.springframework.cloud.st
浏览 10提问于2018-09-08得票数 0
我正在使用KafkaStreams处理这些数据(包括会话控制)。有许多输出主题被发送到数据库。我有一个TimestampExtractor，它将卡夫卡记录的时间戳设置为原始事件的时间戳，以便对数据执行窗口化。但是，在处理的输出主题中，我已经设置了长达几周的保留策略(因此它们在被消耗后就会被删除)。如果我从原始主题重新处理这些数据，在输出主题中生成的时间戳可能会比保留策略的阈值更早--因此它们可能被标记为删除。
既然当它们发表时，它们就有资格保留，我怎样才能防止它们被删除呢？如何<
浏览 1提问于2018-07-19得票数 0
回答已采纳
我们正在编写一个发布到Kafka的新应用程序，我们需要将消息的键序列化为一个长值。在检查Kafka文档进行序列化时，似乎有两个长的序列化类，如LongSerializer和Serdes.LongSerde。我们试图为两者之间的差异寻找一个参考，但没有找到任何解释差异的链接。如果有人能让我们知道/分享一个解释它们之间的区别的链接，非常感谢。或者这些都是一样的？LongSerializer：谢谢。
浏览 1提问于2019-10-01得票数 0
回答已采纳
StreamsBuilder builder = new StreamsBuilder();
);
final KafkaStreamsstreams = new KafkaSt
浏览 2提问于2018-03-15得票数 11
在我的Java应用程序中，我有一个Kafka处理器。我启动应用程序并记录如下：In the process method, the offset is: 1204props.put(StreamsConfig.STATE_DIR_CONFIG, "somedir");下面的代码片段显示了如何创建= new <em
浏览 0提问于2019-04-22得票数 1
回答已采纳
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM
腾讯云开发者
