我有一个混合匹配的Scala拓扑，其中主要工作者是一个PAPI处理器，其他部分通过DSL连接。DoubleKey.getSerializer(), Visitor.getSerializer(), visitorSinkPartitioner, EVENT_PROCESSOR_NAME)
在DSL然而，在我的测试中(使用，但这不会有什么不同)，如果我用一个分区运行它们，一切都很好( EventsProcessor在同一个DoubleKey上的两个事件上检查KTable，在第二个事件上-有
浏览 14提问于2018-07-26得票数 0
回答已采纳
我的卡夫卡主题包含由deviceId键控的状态。我希望使用KStreamBuilder.stream().groupByKey().aggregate(...)只将状态的最新值保存在TimeWindow中。我猜想，只要主题被键划分，聚合函数就可以以这种方式返回最新的值：这是我能从卡夫卡流中得到的保证吗？我应该使用自己的处理方法检查时间戳吗？
浏览 2提问于2017-01-09得票数 6
回答已采纳
我们正在尝试实现下面描述的用例，我们有我们希望克服的实现问题，我们正在尝试通过匹配两个流的消息中存在的关键字(JSON)来在两个Kafka主题之间进行KStream连接。例如，KTable有密钥100，但KStream还没有收到密钥100，那么我们应该重试Join或按住KStream，直到密钥100到达。此外，我们必须从非键控主题构建键控KStream (密钥将通过从Message -JSON中提取来设置)KTable</em
浏览 2提问于2019-05-08得票数 2
我试图通过更改键从流到主题创建一个表，但是值保持不变。是否有可能避免重新划分？ TOPIC, ) .filter { _, value -> value != null }
浏览 3提问于2022-02-07得票数 0
回答已采纳
1回答
我们已经开始对Kafka进行实验，看看它是否可以用于聚合我们的应用程序数据。我认为我们的用例与Kafka流是匹配的，但是我们不确定我们是否正确地使用了这个工具。我们的概念证明是使用kafka流在输出主题中记录有关程序的信息。numberActive": 0, "lastLogin": "01-01-1970T00:00:00Z" 计算计数简单，本质上是根据输入主题和输出字段执行比较和<
浏览 0提问于2018-03-26得票数 1
回答已采纳
最近，我正在为Kafka Streams DSL应用程序实现一个Kryo序列化程序。Kryo在默认情况下不是线程安全的，序列化方法抛出异常很可能是由多个线程的不同步访问引起的。Kafka Streams应用程序针对不同处理对象的线程模型是什么？哪些对象在线程之间共享，哪些对象仅由单个线程使用？在这些对象中使用非同步的本地状态(字段，而不是状态存储)是否安全？请注意，我并不是要在线程之间共享任何状态，而是要避免这种共享状态。DS
浏览 1提问于2020-12-04得票数 0
2回答
一般来说，我有一个关于流媒体的问题，但为了这个问题的范围，让我们用Kafka Streams来限制自己。让我们进一步缩小范围，将我们的问题限制为只计算字数，或者可能计算一般的字数。假设我有一个由一些键和一个值组成的流，key可以是一个字符串(假设我们可以有许多字符串，除了空字符串，由世界上任何字符组成)，而value是一个整数，现在我们正在构建一个单词统计应用程序，如果词汇表中的单词总数是万亿我无法在KTable或任何其他本地存储中存储万亿个单词，我该如何构建此应用程序？我对流的理解是错误的，或者它们是如
浏览 3提问于2019-04-18得票数 1
2回答
我的理解是kafka流支持分区。我想知道，当连接两个不同主题的数据时，它是如何工作的？我假设，为了根据两个不同的主题连接数据，客户端应用程序必须在一定程度上保证来自两个主题的消息共享相同的密钥。
浏览 2提问于2020-08-01得票数 3
回答已采纳
1回答
如我所想，KTable -带有compaction删除策略的简单kafka主题。此外，如果为KTable启用了日志记录，则还会有changelog，然后，删除策略为compaction,delete。KStream<K, V> source = builder.stream(topic1);<em
浏览 2提问于2018-09-25得票数 19
回答已采纳
我在Kafka流中是新手，我的用例是将来自输入主题的每个单独传入记录的值与来自其单独先前记录的值进行比较，如果比较条件为真，则将包含比较结果的新记录与每个比较记录的索引发送到结果主题，否则不发送任何内容在Kafka消费者和生产者API中做这件事非常容易，但是((不使用外部DB来存储前面的记录)我只尝试使用(Kafka streams DSL API)，(其中包括KTable和KStream，以及它们的内部方法等)，但可能是因为我是一个初学者，所以我没有
浏览 2提问于2018-12-23得票数 1
在使用连接到Kafka的Spring应用程序中，我尝试设置两个单独的流侦听器方法：
at org.apache.kafka
浏览 4提问于2018-10-15得票数 1
回答已采纳
最近，在一次采访中，我被问到一个关于Kafka Streams的问题，更具体地说，面试官想知道为什么/什么时候你会使用Kafka Streams DSL而不是普通的Kafka Consumer API来读取和处理消息流
浏览 2提问于2020-09-13得票数 3
Kafka在0.10版本推出了Stream API，提供了对存储在Kafka内的数据进行流式处理和分析的能力。小白一名，各位大佬能给我解释一下流式计算、Kafka Streams的特点和架构吗？谢谢！
浏览 2300提问于2018-09-14
为此，我为此创建了一个转换器，创建了一个流A'，并使窗口意识到了它(也就是说，现在，S将由A'而不是A生成)。A -> A' --> [S]但是我不能这样做，因为当我创建拓扑时，会抛出一个异常： }
val sessionStore : KTable</
浏览 0提问于2019-05-08得票数 1
1回答
我想知道是否有任何方法可以使用Kafka流、DSL或处理器API对窗口中的记录进行排序。假设主题有10个分区。消息由键进行分区，但每个键都是唯一的，因此它是一种随机分区。每个记录都包含一个用户id，该id稍后用于重新划分流。我理解Kafka流的窗口模型，当新记录出现时，时间会提前，但我需要这个窗口使用处理时间，而不是事件时间，然后当窗口过期时，我需要能够对缓冲事件进行排序，并按照这个顺序将它们发送到另一个主题。注意：
我们需要能够使用处理时间<em
浏览 2提问于2018-03-13得票数 7
我知道kafka streams提供的并行性优势，如果您的并行性需要与主题的分区相一致，就可以利用这些优势。我之所以考虑不同的线程，是因为我想要多个动态创建的流的KTable实例。每个流都在相同的流(而不是子集)上工作，并聚合不同的结果。我相信每个线程都可以在自己的streams实例(和使用者组)上工作的想法是有效的。在线程被用于使用某个主题的情况下，让线程遵循一般的kafka并行概念，即处理数据的不同子集，从而使线程并行性成为一个类
浏览 1提问于2017-05-02得票数 1
2回答
我正在导入一个DB，它包含一些表示多到多关系和一对多关系的链接表。最后，我需要将这2与它的所有文档一起加入到一个完整的BioAssay中，例如BioAssayxyz，.到目前为止，根据我对Kafka流的理解，为了执行聚合，我似乎需要为每个链接表提供一个流。KTable将无法使用，因为每个键都会更新记录。但是，聚合的结果可
浏览 0提问于2019-07-07得票数 0
回答已采纳
3回答
消费者和流的区别是什么？对我来说，如果有任何工具/应用程序使用来自Kafka的消息，那么在Kafka世界中，它就是一个消费者。
流是如何不同的，因为这也消耗或产生信息给卡夫卡？
浏览 2提问于2017-05-17得票数 158
回答已采纳
假设我有一个有许多分区的主题。我在里面写K/V数据，并希望在中按键将所述数据聚合到Windows中。
假设我已经启动了尽可能多的工作实例，并且每个辅助实例都运行在一台单独的机器上。
浏览 3提问于2016-08-31得票数 5
回答已采纳
我能找到的最好的例子是@Input和@Output通道都已经是KStreams，但我认为这不是我想要的。从那里开始，我希望得到它的输出，并在KStream字段上构建一个KTable和一个platformUID键。(这里我以字符串形式读取时间，而不是在java/kotlin实现中所做的实际时间) platformUID string, bootstrap-servers: # This
浏览 1提问于2019-08-02得票数 0
云服务器
ICP备案
对象存储
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号