每个主题都有密钥和有效负载。我尝试加入前两个主题，汇总结果，最后将这个结果与第三个主题结合起来。但它并不像预期的那样起作用。我尝试将“映射”与“类别”连接起来，并对结果进行聚合"name“。此操作失败，引发以下错误：处理器KTABLE连接-输出-0000000018不能访问StateStore
浏览 1提问于2020-02-04得票数 1
摘要我要它使用Kotlinx序列化服务器。我如何具体说明这一点？详细信息按块分组瓦
首先，我将瓷砖分组为32x32块，然后将其分组到一个KTable中。tilePrototypesTable: KTable<SurfaceIndex, SurfaceProt
浏览 14提问于2022-01-20得票数 0
回答已采纳
我正在写一个使用Kafka streams的应用程序。它从主题A读取，进行一些转换，并写入主题B。在转换期间，值按键分组，因此输出键、值类型与输入值类型不同。Kafka streams使用特定类型的Serdes (例如String serdes序列化和反序列化字符串)进行序列化和反序列化，因此在数据转换后它将不起作用。如何在Streams API中定义不同的序列化程序和反序列化程序？
浏览 15提问于2018-07-27得票数 0
回答已采纳
1回答
我在我的项目中使用Kafka stream 2.3.0依赖项，但当我使用2.3.0构建JAR时，我看到2.0.1 Kafka版本并导致Kafka连接问题。我从来没有在我的代码中使用2.0.1，我想知道Kafka从哪里获得这个版本 <streams.version>2.3.0</streams.version><dependency&g
浏览 2提问于2020-08-13得票数 0
我正在运行一个Kafka应用程序，它使用来自2个主题的数据，并将连接/合并结果输出到3个主题。kafka主题有15个分区和3个复制因子。我们有5位kafka经纪人和5位动物园管理员。:49) at org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:268) at org.apache.kafka.streams.proc
浏览 4提问于2017-08-09得票数 6
我需要写一个java应用程序，从Kinesis流读取和存储数据到Kafka主题。我不能使用连接器，我必须使用Kafka Streams，因为其中涉及到很多逻辑。Kafka Stream是否可以直接从Kinesis流读取，或者我必须先将数据存储到Kafka topic才能被Kafka Stream读取？我读了很多关于Kafka Streams的文章，除了Kafka主题之外，我从来没有见过Kaf
浏览 18提问于2019-07-29得票数 1
我使用的是带有查询模式的jdbc源连接器，似乎没有指定表名的情况下，在模式注册表中为记录键和记录值注册的模式具有空的模式名称，并被分配了默认名称"ConnectDefault“，如Confluent的AvroData类中所定义的那样Exception in thread "streams-app-6e39ebfd-db14-49bc-834f-a
浏览 4提问于2017-11-03得票数 2
当kafka-streams应用程序运行，Kafka突然关闭时，应用程序进入“等待”模式，用户和生产者线程发送警告日志给他们无法连接，当Kafka回来时，一切都应该(理论上)恢复正常。再一次，和setUncaughtExceptionHandler一样，这里也没有异常抛出，所以什么也没有发生。
我知道卡夫卡有自己的衡量标准，我可以听，如果经纪人被打倒了，我可以找到。但是，在有些情况下，Kafka代理很好，而我的kafka-st
浏览 2提问于2020-08-23得票数 5
回答已采纳
1回答
我有一个按键分组的流，然后我试图像这样对它进行窗口化： groupedStream.windowedBy:73) at org.apache.kafka.streams.state.inter
浏览 7提问于2020-05-19得票数 0
回答已采纳
1回答
我只是在学习Apache Kafka。我当前的ETL在batch process上运行，现在我希望它在流进程上运行，以便用于报告的数据始终是最新的。据我所知，我可以使用MongoDB连接器来捕获mongodb中的数据变化，然后将其发送到kafka主题。但在我的ETL中，我需要将处理后的数据存储到SQL数据库中。我是否可以使用AWS lambda函数来执行处理和记录创建？但是，我如何在kafka中调用这个函数呢？
浏览 4提问于2020-06-17得票数 0
我正在使用Java1.6运行我的应用程序，并计划将一些方面与Kafka集成。
我正在探索是否应该使用Kafka Consumer (使用poll())或使用KafkaStreams API。
浏览 3提问于2018-12-20得票数 0
我已经在kafka streams上工作了很长一段时间。我被困在一个问题中，无法解决它。我期待着来自这个平台的一些帮助。所以场景是这样的。有许多客户和许多为这些客户工作的员工。我想要流式传输该主题，并希望以同一客户下的所有员工分组的方式构建一个物质化商店。我使用开源的ListSerializer和ListDeserializer类来完成这项工作。
所以主题定义是这样的。它从存储中读取主题和流数据，将customer-id作为关键字，将员工列表作为关键字的值。我已经将ArrayListtS
浏览 2提问于2020-05-16得票数 1
使用汇合5.4.1外键跨ktable连接之后，虽然使用“for each”从ktable消费流，但如果使用“似乎卡夫卡2.4.0已经有了类似的问题：和这个问题继续存在，同时将它转发到另一个主题
Caused by: io.confluent.kafka.schemar
浏览 5提问于2020-03-16得票数 0
回答已采纳
我有一个KStream x KStream连接，它将按照下面的异常进行分解。-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_15, processor我将加入一个Click主题和一个Recommendation主题。Click对象非常小(小于一个KB)。另一方面，Recommendation可能很大，偶尔大于1MB。执行连接的代码如下所示。= new Kafka
浏览 3提问于2017-10-05得票数 9
1回答
我需要从主题创建一个Ktable，其中键(metadata.entity_id)和值作为元数据。该表稍后将用于与具有相同键的流进行连接。kafka-streams 2.2.0 Exception in thread "TestMetadataTable-StreamThread-1" org.apache.kafka.streams.errors.ProcessorStateExceptionat org.apache.kafka.strea
浏览 34提问于2020-04-29得票数 1
回答已采纳
在Scala中，我正在尝试一个非常简单的Ktable到Ktable的连接。~[kafka-streams-2.4.0.jar:?]~[kafka-streams-2.4.0.jar:?]at org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:536) [kafka-streamsat org.apache.kafka</
浏览 0提问于2020-03-09得票数 1
我的源是MySql，我使用debezium连接器将其连接到Kafka connect。来自源的数据格式为JSON。现在在KSQL中，我已经创建了一个派生主题，并将JSON主题转换为AVRO，以便可以使用JDBC连接器将数据转换到MYSQL。(PartitionGroup.java:117) org.apache.kafka.streams.processor.internals.StreamTask.addRecords(:831) org.apache.
浏览 26提问于2019-07-10得票数 0
是将ProducerConfig和ConsumerConfig值放入spring.kafka.streams.properties，还是通过spring.kafka.producer和spring.kafka.consumer目前，我是通过直接在代码中设置ProducerConfig和ConsumerConfig值来做到这一点的。如果我通过ProducerConfig和ConsumerConfig提供spring.kafka.
浏览 1提问于2019-06-13得票数 3
回答已采纳
我使用的是合流模式注册表和Avro。使用JDBC连接器将数据输入kafka，后者使用SMT来创建正确的avro模式。使用SpecificAvroSerde反序列化时会出现问题。因此，通常使用avro在流处理器中摄取数据、生成avro模式和使用流处理器的方法是可行的。与本例不同的是，该记录包含一个数组(某种主/详细记录)。连接器和sonsumer都使用相同的avro jar版本。(StreamThread.java:914) ~[kafka-stream
浏览 5提问于2022-06-07得票数 0
回答已采纳
云服务器
ICP备案
云直播
即时通信 IM
对象存储
腾讯云开发者
