我希望从KStream实现一个KTable，并且希望该KeyValueStore按键排序。我尝试查找排序API规范(https://kafka.apache.org/20/javadoc/org/apache/kafka/streams/kstream/KTable.html)，但没有‘KTable我还查阅了这篇文章(https://dzone.com
浏览 37提问于2019-04-11得票数 1
回答已采纳
1回答
我有一个流拓扑，它从一个主题中消耗，运行一个聚合，并构建一个KTable，并将其具体化为rocksDB。我有另一个应用程序，它每天消耗来自同一主题的所有事件，并为满足某些特定条件的事件发送墓碑消息(即不再需要它们)。聚合处理这个问题，并从状态存储中删除，但我正在监视状态存储的大小或更改日志主题--任何真正告诉我ktable大小的内容。
我已经公开了JMX指标，但是没有什么能给我提供我所需要的。我可以看到rocksDB中"pu
浏览 0提问于2020-04-01得票数 2
回答已采纳
1回答
试图通过KTable从“连接-信任”主题中获取记录 view = streams.store(ktable.queryableStoreName(), QueryableStoreTypes.keyValueStore为什么哪里没有查询的商店？我看到了像"test_connect-configs_12-connect-configsSTATE-STORE-0000000
浏览 0提问于2018-04-18得票数 3
回答已采纳
我是Apache的新手，我的理解是，当使用KTable时，您只得到对记录的最后更新。不过，我要拿回所有匹配的记录。KTable<String, String> myTable = builder.table(topicName, Consumed.with(AutoOffsetReset.EARLIEST);
kafkaStreams.start();
Thread.sleep
浏览 3提问于2022-11-22得票数 1
回答已采纳
我使用的是处理器API (PAPI)拓扑。val builder = new StreamsBuilder()我得到了一个KTable，但是拓扑只允许您在StoreBuilder中使用，而不是KTable本身。key
浏览 0提问于2018-06-15得票数 2
回答已采纳
我正在尝试编写我的第一个练习应用程序，使用Kafka Streams来计算主题中的字数。但是，我想我指的是旧的API，因为在lambda函数的末尾，我想将KTable的输出放到一个主题中，但我没有看到任何这样的方法。 我引用的代码使用了to()方法，但我认为现在没有这样的方法。我看到了toStream()，但不知道如何使用它将消息放到特定的输出主题中。 有没有人可以看一下，因为这应该是非常基础<em
浏览 19提问于2019-12-26得票数 2
2回答
我们计划在两台机器中运行kafka流应用程序。每个实例都将其Ktable数据存储在自己的机器上。我们现在面临的挑战是，
假设在每个实例中存储了500 K记录。不可能在单个get over http中获取来自其他实例的所有记录(除非有任何流TCP技术可用)。基本上，我们需要一个调用中
浏览 6提问于2017-07-14得票数 1
回答已采纳
1回答
这是我的应用程序，它简单地从客户主题(输入绑定)获取对KStream的引用，从order主题获取另一个引用(订单绑定)。然后，它从客户主题创建一个KTable，并与order KStream执行连接：class ShippingKStreamConfiguration {
浏览 1提问于2019-04-17得票数 0
回答已采纳
我使用的是Kafka Streams 0.10.1.1版本。然而，当我实现这一点时：
val stateStore = Stores.create(stateStoreName).withStringKeys().withStringKeysorg.apache.kafka.streams.kstream.internals.KStreamAgg
浏览 1提问于2017-01-23得票数 2
我正在处理一个Kafka应用程序，我在弄清楚如何使聚合工作时遇到了一些困难。config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, jsonSerde.getClass().getName());KTable<String, Long> total
浏览 0提问于2018-11-13得票数 6
回答已采纳
我有一个带有一些金融货币引号的KTable，它是这样创建的：public KTable<String, Quote> quoteKTable，但是它在foreach()中没有输出任何值，就好像表是空的一样。我开始认为我不能像这样注入KTable bean，但是关于kafka主题的Spring文档非常稀少，我找不到好的例子。任何帮助都将不
浏览 0提问于2018-09-06得票数 1
> input();这在反序列化错误中失败，在调试之后，我看到原因是消息的内容类型标头被设置为application/json，因此它试图将消息反序列化为JSON。内容类型标头是如何设置的，我如何覆盖它？我不是以JSON的形式发送消息，而是像常规的Kafka消息那样以byte[]的形式发送消息，所以我希望使用我的自定义反/序列化程序
更奇怪的是，如果我将它更改为KTab
浏览 0提问于2019-07-15得票数 1
回答已采纳
1回答
我有一个特殊的情况，我需要确保parent键的所有消息都转到特定的分区。在此之前，我已经实现了其他分区，但我总是能够从关键部分获取相关信息。接口显示方法分区的可能性键和值。但是，它也表示它们都可以是空的。KTab
浏览 5提问于2022-06-13得票数 0
回答已采纳
1回答
1.)现在我希望streams应用程序(应该是一个在容器中运行的jar )只转换我想要的形状的数据。这里遵循物化的视图范例。 streams.start();
2.)和另一个流应用程序(应该是运行在容器中的jar )将KTable保存为某种存储库，可以通过包装rest服务访问。在这里，我被困
浏览 3提问于2017-08-11得票数 1
我有一个日志压缩的主题卡夫卡，我正在阅读在一个KTable和创建一个商店。每当我更新一条键控信息时，我都不会看到应用程序运行时的最新状态。但是，在重新启动streams应用程序时，我能够看到更新的状态。如何在不重新启动的情况下运行流应用程序时获取更新的状态。 try {
metaTableView = waitUntil
浏览 1提问于2018-06-19得票数 0
回答已采纳
我正在将来自卡夫卡主题的消息存储在KeyValueStore中，以便以后查询它们。我创建一个KTable如下所示：我将application.yml中的使用者配置为：spring.cloud.stream.k
浏览 1提问于2019-07-03得票数 1
回答已采纳
1回答
我有一个主题timeOffs2，它包含带有键timeOffID的时间间隔信息，值的类型是包含employeeId的对象。我只想将员工密钥的所有时间间隔分组，然后写信给商店。StreamsBuilder builder = new StreamsBuilder();
KTableDeserializer<ArrayList<TimeOff>> deseri
浏览 3提问于2019-10-07得票数 0
回答已采纳
我正在使用一个KTable实现这个主题，并使用它的存储来运行交互式查询：
KTable<Integer, Customer> customers = builder.table(Topics.CUSTOMER.keySerde这对于恢复不起作用，因为压缩的主题接受一个中间消息，在我的例子中，这个消息不能包含客户的全部信息(它可能是一个带有部分信息的更新事件)。根据KStreamBuilder.table的javadoc，创
浏览 8提问于2018-08-21得票数 2
回答已采纳
下面是我所看到的日志( Processing:...部件是从值合并器中记录的)：Event:Event2Updated中创建条目，因此状态是否为空并不重要。尽管Event2需要将它的更改应用到现有状态，但是它没有找到任何更改，因为第一个状态突变仍然没有写入到KTable (它仍然没有被to方法处理)
无论如何，是为了确保我的leftJoin和写入
浏览 0提问于2017-09-14得票数 5
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
