apache-kafka-streams 相关问题

如何在flink中实现KGroupTable的应用案例？

我正在flink上做一些poc，但我找不到关于如何在kafka流中实现类似KGroupTable的用例的文档，如下KTable所示。 ...

在Kafka中添加重新分区后，丢失旧的聚合记录。

我有一个Kafka流，操作如下：stream.mapValues.groupByKey.aggregation。Aggregation基本上是将记录添加到列表中。现在我把实现改成了：stream.flatMap...。

由于kafka流存储等待运行，应用程序无法启动。

我有一个使用kafka流的spring boot应用（kafka docker image: wurstmeisterkafka:2.12-2.1.1, kafka dependencies: org.apache.kafka:kafka-streams:2.4.1）。在应用程序启动时...

如何创建一个以集合对象为值的KStream？

我在Spring云流kafka流binder上工作。在我的消费者Bean方法中，我想返回以List of String为值的KStream--@Bean public Function , ...

在kafka流媒体中用另一个对象替换json对象。

我正在研究Kafka Streams。我正面临以下问题。详细介绍一下我目前所做的工作。我创建了以下主题、流和表：.kafka-topics --create --zookeeper localhost:2181 --。

Spring Cloud Stream应用程序中的timestampExtractorBeanName设置不能覆盖默认值。

我的 Spring Cloud Stream 应用程序使用 Kafka Streams Binder，有以下属性： spring.cloud.stream.bindings: windowStream-in-0: destination: input windowStream-out-0: ...

Kafka streams Exception: org.apache.kafka.streams.errors.StreamsException - Deserialization异常处理程序。

我正试图用kafka流实现一个简单的计数器。它接受一个键值，如果相同的键值来了，它就会添加新的值。这是我目前写的代码 package exercises;import org.apache...。

Kafka流API--反序列化dynyamically生成的JSON。

我一直在尝试寻找以下情况的解决方案，但没有结果。我有一个Kafka Streams应用程序，它应该从一个单一的输入主题中读取一系列JSON对象，所有的对象都不完全... ...

如何在Spring Cloud Stream App中反序列化MessagePack？

我正试图使用Spring Cloud Stream创建一个Kafka流应用，但在反序列化输入消息时遇到了困难，因为这些消息的值已经使用MessagePack进行了编码。如果有人能帮忙，我将会...

使用groupByKeywindowedBycount对Kafka流进行单元测试。

我的问题类似于 如何对一个使用会话窗口的kafka流应用进行单元测试 拓扑结构看起来像 .filter() .groupByKey() .windowedBy(SessionWindows.with(30).grace(5)) .count() ......

KStream的问题，以avro格式读取前后的有效载荷。

我正在汇流kafka平台上开发KStream应用。数据在源主题上以嵌套的avro格式存在，如下所述{ "type": "record", "namespace": "xyz", "table": "abc",...。

如何在Kafka主题中写KTable？就像我们在KStreams中使用 "to() "一样，如何在KTable中写？

由于KTable没有 "to() "方法，所以我们是否需要在发送任何消息到主题之前将其转换为KStream？或者说，我们如何在话题中存储一个KTable？

卡夫卡流重试PAPI和死信。

我试图在kafka流处理器拓扑中实现一个重试逻辑，在有一个异常生产到sink主题的情况下。我正在使用一个自定义的ProductionExceptionHandler来...

消费者作业与Kafka流的多个主题。

如果这个问题之前已经在这里讲过了，请原谅，我找不到任何与之密切相关的内容。我有这个Kafka Streams应用，它从多个topic中读取记录，将记录持久化在DB上，然后 ...

Spark Kafka Producer抛出太多打开的文件 Exception

我正在尝试运行一个用Java编写的Spark Kafka Job，以产生大约10K记录，每批到一个Kafka Topic。这是一个Spark批处理作业，它读取100个(共100万条记录)hdfs部分文件... ...

如何从KT表中删除旧键

我有一个生产者，可以将键为A或B的记录发布到Kafka主题中。在一个流应用中，我将每个键为A的记录平铺到键为U、V或W的记录上，而每个键为...

kafka流状态存储问题与max.request.size参数的问题。

我们在项目中使用Kafka流状态存储，我们想存储超过1MB的数据，但我们得到了以下异常。消息序列化后是1760923字节，大于... ...

使用TopologyTestDriver和Micronaut对Kafka流进行单元测试。

当我试图运行一个基本的单元测试时，我得到以下错误：java.lang.IllegalArgumentException: Unknown topic: streams-wordcount-output at org.apache.kafka.streams.TopologyTestDriver...。

无法解析符号 "StreamsBuilder"。

我试图用Java创建一个Kafka Streams应用程序，但在管理依赖关系方面遇到了困难。请允许我说，我对Java和它的工具生态系统完全是个新手。在...

如何使用自定义的Kafka Streams ProductionExceptionHandler？

我正在开发一个kafka-streams应用程序，但在覆盖默认的ProductionExceptionHandler方面遇到了麻烦。我已经创建了一个实现ProductionExceptionHandler的类。