获取KStream的最新记录

文章/答案/技术大牛

发布

1回答

apache-kafka、apache-kafka-streams、spring-kafka

我有一个这样的KStream： KStream<Long,String> joinStream = builder.stream(("output")); 带有记录数的KStream值如下所示：但我一直在努力获取最后一张唱片的价值。

浏览 20提问于2021-03-29得票数 1

1回答

如何在Hashmap中存储KStream的对象类型的值？

java、apache-kafka、hashmap、apache-kafka-streams、spring-kafka

我有一个像KStream<String, Object>这样的KStream对象。我想在HashMap中获取对象的值，并在不同的KStream中处理后将其存储为值。例如,KStream值= {id=1，name="abc“age=12} 期望的结果: HashMap hm ={id=1，name="abc“age=12}

浏览 0提问于2020-05-19得票数 0

1回答

定制窗口商店在KStream - KStream join中的实现

apache-kafka、apache-kafka-streams

我们需要使用一个非常大的窗口执行Kstream - Kstream连接，其中左侧的勾号只会触发与右侧最新记录的连接，反之亦然。这不是默认窗口的工作方式，因为window.fetch在KStreamKStreamJoinProcessor中返回的WindowStoreIterator<V>是一个迭代器，可以包含多条记录。特别是，我们注意到RockDBWindowStore的retainDu

浏览 1提问于2018-05-23得票数 0

1回答

如何在不改变主题的情况下加入KStream和KTable

java、apache-kafka、apache-kafka-streams

我加入了KStream和KTable。下面是我的密码。builder = new StreamsBuilder();KStreamconsumer_offsetsaddress如何在没有“changlog”主题的情况下执行代码

浏览 4提问于2021-04-09得票数 0

3回答

为什么要使用KStream或KTable？

apache-kafka、apache-kafka-streams

或者它有更多的功能。然而，在这方面，我找不到一个好的例子。我在一个很好的解释工作逻辑的来源中也看不到它。你能解释一个很好的ktable和kstream的例子吗?我能做些什么？

浏览 0提问于2020-03-01得票数 0

2回答

如何将记录分成不同的流，从一个主题到不同的流？

apache-kafka、apache-kafka-streams、apache-kafka-connect

我有一个源CSV文件，包含不同大小的记录，将每条记录推入一个源主题。我想把记录分成不同的KStreams/KTables与那个源主题。我有一个用于一个表加载的管道，在该管道中，我以分隔格式将记录从源主题推入stream1，然后将记录按AVRO格式推到另一个流中，然后将该流推入JDBC接收器连接器，该连接器将记录推入MySQL数据库。管道应该是一样的。但我希望将不同表的记录

浏览 1提问于2020-04-13得票数 0

回答已采纳

2回答

具有相同密钥的KStream leftJoin KStream

java、apache-kafka、left-join、apache-kafka-streams

我在尝试加入两个流的时候遇到了问题。mergedKey有相同密钥的100+列表，而DataStream只有1个列表与mergedKey具有相同的密钥。//get DataStream // Transform merged to Equals Keys to

浏览 15提问于2018-08-25得票数 0

1回答

Kafka流: ktable作为查找和目标合并流

java、apache-kafka、apache-kafka-streams

我找不到解决并发问题的简单解决方案： KTable<String, ProdForecastPowerPlantAgg> mergedTable= recordsAStream KStream= recordsBStream .sel

浏览 0提问于2019-05-21得票数 2

回答已采纳

1回答

Kafka流可以将输出写入两个不同的主题吗？

apache-kafka、apache-kafka-streams

我需要一个流来对主题A中的值进行分组，将分组后的值发送到主题B，并通过翻滚时间窗口将这些分组的值的总和发送到主题C。在Kafka中可以做到这一点吗？或者一个流只能从一个主题中读取和写入？

浏览 6提问于2018-08-08得票数 1

回答已采纳

1回答

我有一个主题中的数据，需要在多个级别进行统计，所有代码和文章都只提到字数统计示例。数据的一个例子是：序列号: 123国家:美国日期:2018年5月1日州:纽约市:纽约市访客人数:5 序列号: 123国家:美国日期:2018年6月1日州:纽约市:皇后区游客: 10 序列号: 456日期// }Materialized.with(stringSerde, visitorSerde)); 对于所有相等的serial_id(这将是group by)，计算每次访问

浏览 10提问于2019-06-17得票数 1

回答已采纳

1回答

kafka streams +如何异步终止状态存储中的条目

apache-kafka、apache-kafka-streams

如果它可以被删除，它将被删除，否则我有一个标点符号，它每10秒运行一次，并使状态存储中的项过期。感谢你的帮助。

浏览 10提问于2020-02-08得票数 2

2回答

卡夫卡蒸汽应用程序是否有可能从一个输入中编写多个输出？

apache-kafka-streams

我不确定卡夫卡流是否是我想要解决的问题的正确解决方案。我希望能够使用它，因为它提供了并行性和容错性，但我很难找到一种方法来实现所需的处理流水线。我希望能够将每个结果写成一个单独的记录，有它自己的键，而不是作为一个记录中的结果集合。暂时忽略每个结果要求<

浏览 0提问于2018-11-17得票数 1

回答已采纳

1回答

如何将卡夫卡主题中的多个事件组合为新事件

apache-kafka、ksqldb

我有一个关于Json数据的Kafka主题，如{"storeid": 17, "data": "blublublu"}我需要的是将新事件发布到另一个主题/流中，当1000个事件或5分钟(这是第一次)传递时，包含相同存储区<e

浏览 4提问于2022-06-26得票数 0

2回答

卡夫卡流reduceByKey对leftJoin

apache-kafka、apache-kafka-streams

乍一看，在我看来，使用KStream#reduceByKey可以实现与KStream to KTable leftJoin相同的功能。也就是用相同的键组合记录。我有什么区别，两者之间，也在性能方面？

浏览 0提问于2016-09-11得票数 2

回答已采纳

2回答

如何管理Kafka KStream到Kstream的窗口连接？

apache-kafka、apache-kafka-streams

基于 KStream-to-KStream Joins are always windowed joins，我的问题是如何控制窗口的大小？在主题上保存数据的大小相同吗？有没有什么好的例子来展示开窗口的KStream- to -kStream窗口连接？在我的例子中，假设我有2个KStream，kstream1和kstream2，我希望能够加入10天的</em

浏览 1提问于2017-01-18得票数 9

回答已采纳

1回答

KStream到KTable

java、apache-kafka、apache-kafka-streams

@StreamListener("input")public KStream<?, MyObject> process(KStream<Object, IncomingObject> input) {return table.toStream();我试图将KStream

浏览 0提问于2018-07-23得票数 0

2回答

Kafka流以特定的键连接作为输入。

apache-kafka、apache-kafka-streams、kafka-join

我有3个不同的主题，在模式注册表中有3个Avro文件，我想将这些主题流到一起，并将它们写入到一个主题中。问题是，我想加入的键与我将数据写入每个主题的键不同。现在，我想加入这三个主题，并获得一个新的记录，并将它写入一个新的主题。问题是，我想加入、告警、和事件，基于alarm_id和，并加入基于ne_id的告警和维护。我想避免创建一个新的主题和重新分配一个新的键。在我加入的时候，我是否指定了密钥？

浏览 0提问于2017-01-23得票数 2

回答已采纳

2回答

在Kafka流中加入外键

apache-kafka、apache-kafka-streams

假设我有三个Kafka主题，其中充满了代表不同集合中发生的业务事件的事件(事件源应用程序)。这些事件允许生成具有以下属性的聚合：通过将事件按KTable分组，为产品创建一个productId。KTable以productId作为密钥。没

浏览 0提问于2018-11-12得票数 8

1回答

Kafka Streams存储最新值组合表

apache-kafka、apache-kafka-streams

我正在尝试将值出现的最新组合存储在表中。，每个传感器的最新测量时间。"); Topic.Message.getKafkaName(),return measurementTime; return partitionTime;); KStreamwithKeySe

浏览 6提问于2021-07-06得票数 0

1回答

Kafka流是如何分配分区的？

apache-kafka-streams

我有一个卡夫卡流应用程序，它接收来自主题-1作为KStream和主题-2作为KTable的数据。这两个主题都有4个分区。假设我有4个应用程序实例正在运行，那么每个实例将从主题-1的单个分区接收数据。如果这两个主题都是相同的，那么我猜Kafka流将确保为应用程序分配相同的分区。如果Top-2没有任何键，而是应用程序将从值本身推断出这一点，那么这意味着所有实例都需要从Top-2获取所有分区。卡夫卡流是如何处理这种情况的？谢谢!

浏览 0提问于2018-04-27得票数 5

回答已采纳

点击加载更多