, "state" : "zzz"} (包含订单完整信息的JSON)KStream<Integer, JsonNode> stream = kStreamBuilder.stream(in
浏览 8提问于2017-07-17得票数 1
回答已采纳
越来越多的文章将kafka作为事件存储，并将其用于使用cqrs和事件源构建的应用程序中。如何在kafka (作为事件存储库)查询某个聚合的事件，以完成写入端的操作？
浏览 1提问于2016-10-15得票数 4
我有一个Spring boot Kafka Stream应用程序，它处理所有传入的事件，并将其存储在Kafka Streams内部提供的State store中，并使用交互式查询服务进行查询。在所有这些使用" RocksDB“的Kafka流中，我想用任何其他可以配置的数据库替换这个RocksDB，比如MariaDB或MongoDB。有办法做到这一点吗？如果不是</
浏览 21提问于2020-08-27得票数 4
ROWTIME":123456, "id":1, "person_id":"1234"在这个kafka表中，我有很多具有不同provider_id的不同数据。我需要从这个kafka表中检索所有不同的person_ids，其中provider=3。我是卡夫卡<em
浏览 0提问于2021-04-28得票数 0
1回答
和以下逻辑:对于一个新事件，检查它的聚合是否存在，如果是，更新它+存储它+发出新聚合，否则生成聚合+存储它+发出。，它只是业务逻辑的变化(如何将一个新事件与聚合状态合并)15:07:23,420 ERROR org.apache.kafka.strea
浏览 0提问于2018-11-22得票数 0
回答已采纳
我想将消息聚合成一个状态，但我也希望在预处理步骤中使用该状态，这样也可以更新其他状态。例如，查看此消息是否更改了状态的某些部分，如果是的话，更新跟踪状态特定部分以及状态本身的主题。我能想象的最好的方法是使用能够访问状态存储的转换器，但是在聚合状态之前就这样做，这样我就可以在状态更新之前看到状态的值。(在可能的情况下更新了额外的状态后，所讨论的转换
浏览 0提问于2018-05-24得票数 3
3回答
我已经研究了Interactive queries和KSQL，但我似乎无法确定是否可以基于键查询特定的记录。另外，KSQL和交互式查询会占用整个主题来执行查询吗？
浏览 30提问于2020-01-02得票数 0
回答已采纳
2回答
我有一堆非唯一的字符串字段和事件时间戳进入卡夫卡的事件。我想要创建这些事件的物化视图，以便我可以查询它们。显示两个日期之间的事件
我看到的所有示例都有流上的聚合、连接或其他转换操作。我找不到一个简单的例子来创建一组事件的视图。我不想执行任何操作，我只想能够查询进入流的原始事件。我使用的是Spring <
浏览 0提问于2019-07-02得票数 1
回答已采纳
1回答
在我的理解中，互动查询被带入画面，以解决当某些事件仍未被处理(隐藏在kafka主题中，尚未被处理)时聚合结果的问题。但是，我们不应该预先计算任何我们需要的东西而不是运行时聚合，这不违背我们的理念吗？以下是我所看到<
浏览 1提问于2019-07-26得票数 1
我想使用 entity 的值来写入与实体值相同的主题(实体的值是固定的，所以如果动态创建主题很难，我就可以静态地创建)。;import org.apache.kafka.streams</em
浏览 3提问于2017-03-25得票数 0
我想要有一个Kafka Streams拓扑，它将有五个独立的KStreams从各自的主题读取数据，并将数据转储到一个大型的整体主题中。接下来，我将创建一个GlobalKTable，它将从这个整体主题中读取内容，并实现一个全局存储，比方说，称为lookupStore。我想让这个物化的全局存储基本上作为其他Kafka Streams应用程序的“查找”表。现在，我想要有许多独立的
浏览 41提问于2020-11-11得票数 0
我正在处理一个Kafka应用程序，我在弄清楚如何使聚合工作时遇到了一些困难。我有一个KStream bankTransactions，其中键是String类型的，值是JsonNode类型的，所以我用
// Definition of the different Serdes usedKTable<String, Long>中，其中键将是相同的，但是值将是从Json提取的</em
浏览 0提问于2018-11-13得票数 6
回答已采纳
我正试图确定Kafka Streams州立商店是否适合这个问题：卡夫卡是一个很适合这个问题的人吗? instance?How ，如何确定每个Kafka流所需的内存内存，，，，，，RocksDB，，，，RocksDB，RocksDB，，，指南，以确
浏览 4提问于2020-02-08得票数 1
回答已采纳
我在一个Docker容器中运行了一个KStreams应用程序，它使用一个持久的键值存储。我的运行时环境是RHEL 7上的Docker 1.13.1。我已经将state.dir配置为/tmp/kafka-streams值(这是默认的)。我的application.id是“我的应用程序”。我的输入主题中有288个分区，这意味着我的
浏览 9提问于2022-06-01得票数 0
我正在尝试使用Spring Cloud Stream创建一个简单的功能bean，它处理来自KStream和GlobalKTable的消息，连接它们，聚合它们，并将结果输出到一个新的流中，但我在正确配置它所需的无需多说，下面是我的方法：public BiFunction<KStream<GenericRecord, GenericRecord>, GlobalKTable<Long, GenericRecordat
浏览 82提问于2020-05-20得票数 1
回答已采纳
我试图使用Kafka流对CDC数据执行KTable-KTable外键连接。我将要读取的数据是Avro格式的，但是它是以与其他行业序列化程序/反序列化器(例如)不兼容的方式序列化的。因为架构标识符存储在标头中。当我设置我的KTables‘Serdes时，我的Kafka应用程序最初运行，但最终失败了，因为它在内部用byte[] serialize(String topic, T data);调用序列化器方法，<
浏览 5提问于2021-11-10得票数 2
回答已采纳
定义了一个自定义存储，在自定义Transformer中使用(参考如下)。不确定，为什么内部主题"test_01-HOUSE-changelog“是用单个分区和单个复制创建的，而不是源分区"test”中的2个分区。这里遗漏了什么？(AbstractTask.java:81)
at org.apache.kafka.streams.processor.internals.StreamT
浏览 2提问于2018-05-15得票数 0
我正在寻找一个合适的数据库来存储大量的时间序列数据，并且支持微秒时间戳。有什么建议吗？希望收到您的回音:-)
祝你好运，朱莉安。
浏览 22提问于2017-01-10得票数 0
我已经在kafka streams上工作了很长一段时间。我被困在一个问题中，无法解决它。我期待着来自这个平台的一些帮助。所以场景是这样的。有许多客户和许多为这些客户工作的员工。只有一个主题存储了所有这些数据。我想要流式传输该主题，并希望以同一客户下的所有员工分组的方式构建一个物质化商店。类，这里的T是对象，它是Employee。它从存储中读取主题和流数据，将customer-id作
浏览 2提问于2020-05-16得票数 1
云服务器
ICP备案
对象存储
即时通信 IM
云直播
腾讯云开发者
