我正在使用kafka处理器API进行一些自定义计算。由于一些复杂的处理，DSL并不是最合适的。流代码如下图所示。outputTopic", stringSerializer, resultSerde.serializer(),
Fill_PROCESSOR); 我需要根据单独主题中的事件从状态存储中清除一些项我找不到正确的方法来连接另一个流，使用处理器API或其他方式来侦听另一个主题中的事件，以便能够触
浏览 13提问于2020-09-17得票数 0
回答已采纳
1回答
我一直在尝试Apache Kafka开源生产者和消费者数据流。我有一个问题，使用1.) KafkaProducer, producer.send和2.)它们不都是用于kafka流数据的吗？如果我只使用KafkaProducer, producer.send，我不能实现数据流吗？
浏览 1提问于2018-03-16得票数 0
我有一个带有通过Spring Cloud映射的两个流处理器的SpringBoot应用程序。每个处理器都有自己的@StreamListener用于不同的主题。一个处理器将聚集的数据写入可四分状态存储。当我通过@Service (服务从状态存储获取聚合数据)获取数据时，我在单元测试中遇到了这个问题。at org.apache.kafka.streams.state.internals.QueryableStoreProvider.getS
浏览 10提问于2018-09-08得票数 0
状态存储是从第一个主题构建的，第二个流读取状态存储。当我启动拓扑时，我看到这些流线程被分配了两个源主题的相同分区(共同分区)。我假设这是因为状态存储被第二个主题流访问。对我来说，第二个主题应该尽可能快地处理。处理第一个主题的延迟是可以的。Is there a chance that Kafka will assign 10 partitions
浏览 5提问于2020-10-05得票数 0
我使用处理器api从状态存储中删除消息。Delete是成功的，我通过使用由kafka键对状态存储的交互式查询来确认，但是它并没有减少本地磁盘上tmp/ kafka -streams目录下的kafka流文件大小。statestore.all().forEachRemaining(keyValue -> {
浏览 7提问于2020-02-12得票数 0
回答已采纳
当我们在Windows 10上运行Kafka时，每5秒就会出现以下错误。
未能将偏移量检查点文件写入C:/tmp/kafka/user/ global /.检查点，用于全局存储：{}。如果操作系统清除了state.dir，当它位于(默认)${java.io.tmpdir}/kafka目录时，可能会发生这种情况。改变state.dir的位置可以解决这个问题。java.nio.file.AccessDeniedException: C:/t
浏览 4提问于2022-03-03得票数 0
回答已采纳
当使用处理器API和KTable函数创建全局addGlobalStore时，结果存储将填充OK。at org.apache.kafka.streams.state.internals.SerializedKeyValueIterator.next(SerializedKeyValueIterator.java如果我使用globalTable函数创建全局状态存储(因此这不是反序列化问题)，同样的主题可以正常工作。状态存储也成功地返回行数
浏览 0提问于2017-09-01得票数 1
回答已采纳
2回答
我有一个用例，我需要从主题中读取数据，然后批量数据(100条记录)，并将批量数据写入特定的文件或外部存储。为此，我计划使用处理器API，并使用kafka支持的状态存储对正在处理的数据进行批处理，一旦批处理大小达到100条记录，就将其写入文件。从状态存储中清除批以创建新的新批。streams只有一次适合这个用例吗？我在设计中读到，如果我们在批量处理数据，那么不推荐这样做，而且大多数关于这方面的文章都说，只有一次才能在消
浏览 10提问于2019-01-10得票数 0
我们在项目中使用Kafka流状态存储，我们希望存储超过1MB的数据，但是我们得到了下面的例外：
topic.max.request.size=50000000
然后应用程序可以正常工作，但是当创建了状态存储内部主题时，但是当Kafka重新启动并且状态存储主题丢失/删除时，它可以正常工作，然
浏览 3提问于2020-06-04得票数 1
我已经将我的kafka实例与org.apache.kafka.streams.processor.StateRestoreListener接口的实现联系在一起。我使用处理器API来定义我的流，并且它有一个状态存储。基本上，流的工作是从输入主题中读取数据，并将数据存储在状态存储中。
我想知道在什么情况下调用这个StateRestoreListener。
浏览 1提问于2020-08-20得票数 0
回答已采纳
是否可以在单个事务中生成Kafka主题并写入状态存储？而不是作为主题消费的一部分启动事务。我也知道Confluent提供的解决方案。
浏览 3提问于2019-03-01得票数 0
我正在做一个POC来演示一些Spring Cloud Kafka功能。我使用的是带有SpringBoot 2.4.1的Java11。我的build.gradle具有以下库 implementation 'org.springframework.cloud:spring-cloud-stream'
implementation 'org.springfram
浏览 0提问于2021-01-30得票数 0
我正在使用带有状态存储的低级别处理器API。直到0.10.0.1，它的工作很好，但我已经升级了卡夫卡流，我得到了下面的错误。at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.timestamp(AbstractProcessorContext.javaat org.apache.kafka.streams.state.internals.StoreChang
浏览 1提问于2017-05-18得票数 7
回答已采纳
据我所知，处理Kafka流的状态在内存、磁盘或Kafka主题中都是本地的，因为所有的输入数据都来自一个分区，其中所有的消息都是由一个定义的值键控的。大多数情况下，计算可以在不知道其他处理器的状态的情况下完成。如果是这样的话，您有另一个Streams实例来计算结果。如图所示：Flink到底在哪里存储它的状态？Flink是否也可以在本地存储状态，或者总是将它们发布到所有实例(任务)？是否可以将Flink配置为
浏览 0提问于2019-02-10得票数 3
为什么卡夫卡流state.dir存储在/tmp/kafka-streams下？我知道我可以通过在流代码中提供状态dir配置来更改路径，如下所示但是改变目录会有什么影响吗？或
我可以在应用程序目录而不是/tmp中配置状态DB吗？
浏览 0提问于2019-03-11得票数 6
我正在使用Java1.6运行我的应用程序，并计划将一些方面与Kafka集成。
我正在探索是否应该使用Kafka Consumer (使用poll())或使用KafkaStreams API。我们需要Java 1.7 / 1.8来使用Stream API吗？
浏览 3提问于2018-12-20得票数 0
我正在使用处理器API在状态存储中执行一些低级处理。重点是，我还需要在存储到存储中后写入主题。如何在Spring Cloud Streams Kafka应用中做到这一点？
浏览 40提问于2020-05-02得票数 0
回答已采纳
为了比较使用RocksDB或内存中的Spring Cloud Stream应用程序(使用Kafka)的性能，我想将RocksDB更改为内存中。不幸的是，我找不到任何解决方案。我已经为DSL API和处理器API ()找到了一个解决方案，但我不确定它是否适用于kafka-streams绑定器。有人能在这个问题上帮我吗？
向马丁致以最美好的问候
浏览 20提问于2019-07-10得票数 0
我在kafka 1.0.0上创建了一个流处理器，它不消耗也不产生任何主题--它只是简单地初始化自己。当我试图运行这个处理器时，我会收到以下错误消息：
Exception in thread "main" org.apache.kafka.streams.errors.BrokerNotFoundExceptionKafka Streams requires broker version 0.10.1.x or higher.最后，我一直在研究
浏览 2提问于2018-01-19得票数 4
我需要能够从Ktable中从一个单独的Stream处理器中删除一个记录。今天，我使用聚合()并传递一个物化状态存储。在一个从“终止”主题中读取的独立处理器中，我希望在.transform()或另一个.aggregate()中查询物化状态存储，并‘删除’那个键/值。每次我尝试从一个单独的流处理器访问物化状态时，它总是告诉我要么存储区没有添加到拓扑中，然后我添加它并再次运行它，然后它告诉我它已经被注册并出错了。由: org.ap
浏览 3提问于2020-03-04得票数 2
云服务器
ICP备案
对象存储
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号