我们使用Kafka Streams需要解决以下问题：3-丰富该事件的完整聚合状态的消息，并将其发送到输出主题
重点是我们不能真的丢失一条消息，它必须始终用最新的聚合状态(我们在消息处理过程中实际评估的状态)来丰富传入的消息。到目前为止
浏览 17提问于2017-02-11得票数 4
回答已采纳
我正在使用Java1.6运行我的应用程序，并计划将一些方面与Kafka集成。
我正在探索是否应该使用Kafka Consumer (使用poll())或使用KafkaStreams API。
浏览 3提问于2018-12-20得票数 0
我一直在使用Kafka Streams，并获得了一些基本的功能，但我在理解Kafka Streams DSL中的process方法时遇到了一些问题。具体地说： 我知道有两种使用Kafka Streams的方法，低级处理API和高级Streams DSL。对于较低的级别，您可以更明确地定义拓扑，命名每个节点等，而流DSL则将大部分抽象出来。所以我的问题是，处理过的数据-Processor的void f
浏览 9提问于2020-01-20得票数 0
回答已采纳
对于我的Kafka流应用程序之一，我需要使用DSL和处理器API的特性。我的流媒体应用程序流聚合之后，我需要向接收器发送一条聚合消息。(MeteredKeyValueStore.java:167) at org.apache.kafka.streams.processor.internals.Pro
浏览 4提问于2016-10-24得票数 6
回答已采纳
Kafka是很常见的。所以很多公司都在使用它。我完全理解Kafka和Spark是如何工作的，我对他们都很有经验。我不理解的是用例。为什么你要把Kafka和Spark一起使用，而不仅仅是Spark呢？在我看来，Kafka的主要用途是作为ETL管道中的中转区，用于实时(流)数据。
我假设有一个数据源集群，数据最初存储在其中。例如，它可以是Vertica、Cassandra、Hadoop等。然后是一个处理集群，它从数据源集群读取数据，并将其写入分布式Kafka日志，这基本上是
浏览 3提问于2019-06-17得票数 3
1回答
我只是在学习Apache Kafka。我当前的ETL在batch process上运行，现在我希望它在流进程上运行，以便用于报告的数据始终是最新的。据我所知，我可以使用MongoDB连接器来捕获mongodb中的数据变化，然后将其发送到kafka主题。但在我的ETL中，我需要将处理后的数据存储到SQL数据库中。如何以及在哪里处理从mongodb发送到主题的数据，然后从该主题创建到另一个数据库的记录？我是否可以使用AWS lambda函数来执行处理和记录创建？但是，我如何在kafka中调用
浏览 4提问于2020-06-17得票数 0
我一直在尝试使用Streams DSL在Kafka中实现滑动窗口，但我无法做到。有没有人能帮我举个代码例子。我想要使用滑动窗口方法聚合特定时间段的值。
我搜索了融合的博客，但没有代码
浏览 3提问于2019-07-16得票数 1
1回答
我正在开发一个java应用程序，从来源接收实时数据(Kafka/Kafka流)。在我的案例研究中，我需要过滤我列表中的名字。这份名单最初有1000万个名字。我的第一个问题是知道什么是最好的方法和/技术，以获得更高效/更快的处理时间。
第二个问题是如何更新或删除列表中的一些名称？
浏览 2提问于2020-01-25得票数 0
1回答
使用kafka-streams，我希望将元素的流S E按某些键K1分组，同时将该键的所有值聚合到一个连接的结果AGG中。这将导致KTable T1。根据聚合结果，应该将该值重新划分为另一个KTable T2，按从聚合结果AGG中提取的键K2分组。因此，聚合的结果应该为下一次重组生成密钥。不是每个键K2的值
我知道聚合的结果是在一段时间后才被转发的，所以我已经尝试将commit.interval.ms降低到1，但没有效果。val finalTable = streamBu
浏览 2提问于2019-09-06得票数 1
回答已采纳
每个主题都有密钥和有效负载。我尝试加入前两个主题，汇总结果，最后将这个结果与第三个主题结合起来。但它并不像预期的那样起作用。我尝试将“映射”与“类别”连接起来，并对结果进行聚合"name“。此操作失败，引发以下错误：处理器KTABLE有什么方法可以使这个连接聚合发生吗？是否有可能有这样的输出：
key, valu
浏览 1提问于2020-02-04得票数 1
我加入了KStream和KTable。下面是我的密码。(order, address) -> order + " send to " + address)streams = new KafkaStreams(builder.build(), props);执行此代码后，将创建一个新主题。/kafka-topics.sh --
浏览 4提问于2021-04-09得票数 0
1回答
由于原始数据需要大量存储，所以我试图聚合这些数据，并创建每小时、每天、每月报告的汇总。有两种方法，我可以想到：
流处理:使用Kafka Streams API来卷起数据，并在Cassandra中摄取汇总的数据。
浏览 0提问于2019-03-25得票数 1
1回答
这些是我的DSL调用来完成我的聚合。
val sessionProcessorStream = builder.stream("collector-prod", Consumed.arguments (org.apache.kafka.streams.kstream.KeyValueMapper[org.apache.kafka.streams.kstream.Windowed[org.apache.kafka.streams.ks
浏览 0提问于2018-04-02得票数 0
回答已采纳
1回答
有一个卡夫卡集群，我从其中消费两个主题，并加入它。由于联接的结果，我对数据库做了一些操作。对DB的所有操作都是异步的，因此它们返回给我一个未来(scala.concurrent.Future，但无论如何它与java.util.concurrent.CompletableFuture相同)。因此，我得到了这样的代码：val secondSource: KTable[String, Obj2]
def saveResult
浏览 3提问于2017-02-15得票数 5
最近，在一次采访中，我被问到一个关于Kafka Streams的问题，更具体地说，面试官想知道为什么/什么时候你会使用Kafka Streams DSL而不是普通的Kafka Consumer API来读取和处理消息流
浏览 2提问于2020-09-13得票数 3
它读取主题外的avro消息，并构造一个聚合数据的状态存储，该数据也是avro类型的。}, Materialized.with(new Serdes.StringSerde(), serdeOutEvent).toStream();该函数能够从主题中读取消息并创建第一个聚合结果:27) at org.apache.kafka.
浏览 23提问于2021-02-27得票数 1
回答已采纳
1回答
我最近正在尝试使用kafka streaming进行一些敏感数据处理。我希望实现的目标是，当敏感数据被加密时，微服务架构的能力不会受到影响，即，紧密耦合的服务和流数据处理。我的问题是，在kafka流中，有没有可能我用一个密钥解密传入的消息，然后用另一个密钥再次加密？我有一个计划，但由于我不熟悉kafka streaming，我不能证明kafka streaming有能力使用Streams DSL处理此功能。有没有人能帮我解决这个问题，最好告诉我Streams</
浏览 34提问于2020-01-14得票数 0
回答已采纳
我使用Kafka stream、DSL和map将KStream<String, JsonNode>转换为KStream<String, String>。在ValueMapper函数中，我简单地返回new ValueMapper("key", "some constant string")，但是在使用KStream.to("some topic")将值发送回Kafka的地方，我得到的结果是添加了双引号。也许是Kafk
浏览 2提问于2018-09-03得票数 3
我是新的卡夫卡流和一些卡在一些基本的字计数程序。卡夫卡流版本=> 2.3.0import java.util._import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.streams{KafkaStreams,St
浏览 1提问于2020-03-21得票数 1
回答已采纳
云服务器
ICP备案
云直播
即时通信 IM
实时音视频
腾讯云开发者
