我有一个关于TimeWindows和Kafka Streams的问题，有些概念真的让我很困惑。现在，当我使用以下语句加载所有事件时.windowedBy(TimeWindows.of(TimeUnit.HOURS.toMillis(1)).until(TimeU
浏览 16提问于2018-12-13得票数 0
回答已采纳
我正在使用Kafka和Kafka流作为的一部分。在我的Kafka应用程序中流动的数据正在通过特定的时间窗口进行聚合和物化：TimeWindows <
浏览 3提问于2019-02-14得票数 3
回答已采纳
例如，我试图使用Kafka流对特定时间范围(24小时)的唯一访问者进行统计，Duration windowSizeDuration = Duration.ofDays(1);但是我找不到任何可用的流</
浏览 1提问于2020-07-04得票数 3
回答已采纳
;import org.apache.kafka.streams.KafkaStreams;import org.apache.kafka.streams.StreamsConfig;import org.apache.kafk
浏览 0提问于2019-10-30得票数 0
是否可以使用Java CompletableFutures在Kafka stream应用程序中执行并行工作？ 我想从1Kafka主题中读取，创建两个窗口计数，1分钟，另一个小时，但它们是并行的。我能够让它工作，但是看一下Kafka流文档，因为KafkaStreams为每个分区分配一个任务，并且它不能超过一个线程，我不确定这个代码是否会有预期的效果。completableFutureOfMinute = CompletableFuture.runAsync(() -> {
inputStream.gro
浏览 38提问于2020-07-25得票数 0
回答已采纳
1回答
我正在尝试聚合该流，以获得开窗流中user_id的计数。流没有键，因此需要从值中获取user_id，并聚合该窗口中的活动用户计数并将其打印到控制台/api。toString(), value)) .countByKey(TimeWindows.ofCannot resolve method 'countByKey(org.apache.<
浏览 2提问于2018-02-06得票数 0
我正在构建以下Kafka流拓扑(伪代码)：g1 = gK.windowedBy(TimeWindows.of("PT1H")).一个流(翻滚？)开窗，另一个没有。流的两个部分在相同的键上工作(除了窗口中间引入的WindowedKey之外)。当消息数
浏览 5提问于2020-03-12得票数 2
回答已采纳
我有数据流作为<字符串，string>事件。我想要获得10分钟时间窗口的事件计数，并输出到另一个主题。StreamsBuilder(); .groupByKey()
.windowedBy(TimeWindows.ofA serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.
浏览 41提问于2020-09-18得票数 0
回答已采纳
我是写在卡夫卡流跳窗口代码，其中minMaxCalculator()计算最小和最大内流，一旦流是按键分组。Records.groupByKey().aggregate(new aggrTestInitilizer(), , TimeWindows.ofadvanceBy(TimeUnit.SECONDS.toMillis(1)),aggrMessageSerde,"aggr-test");
一旦我按键分组，我希望处理为中的所有键并行生成的窗口
浏览 1提问于2017-11-29得票数 0
回答已采纳
1回答
我有一个接收记录的Kafka流，我想根据特定字段连接消息。流中的消息如下所示： Key: 2099 email: tom@emample.com} 预期输出： key: 2099} 我可以让流运行得很好，我只是不确定lamda应该包含什么。> inputStream = builder.stream(INPUT_TOPIC);
inp
浏览 9提问于2019-04-03得票数 1
回答已采纳
我在kafka流中做这个简单的窗口聚合：.groupByKey(/* Serde stuff */) Duration.ofSecondsCACHE_MAX_BYTES_BUFFERING_CONFIG更改为0，并将代码转换为0(只添加了suppress步骤：.groupByKey(/* Serde stuff */)
.windowedBy(TimeWindows.ofSuppressed.
浏览 2提问于2020-12-31得票数 1
回答已采纳
我有一个Kafka应用程序，在这个应用程序中，我从一个主题中读取、进行聚合并在KTable中实现。然后创建一个stream并在流上运行一些逻辑。现在在流处理中，我想使用前面提到的KTable中的一些数据。启动流应用程序后，如何再次访问KTable流？我不想把KTable推向一个新的话题。) .windowedBy(TimeWindows
浏览 2提问于2020-07-01得票数 1
回答已采纳
是否可以通过抑制中间结果来获得卡夫卡流中的。 builder.stream<String,Double>(inputTopic) .windowedBy(TimeWindows.offlush state store KSTREAM-AGGREGATE-STATE-STORE-0000000001:
java.lang.ClassCastException: org.apache.ka
浏览 0提问于2019-01-09得票数 6
回答已采纳
1回答
我正在窗口从流环境传入的对象，收集和打印它。使用卡夫卡压制，以避免中间结果。import org.apache.kafka.common.serialization.Serdes;import org.apache.kafka</e
浏览 5提问于2019-11-05得票数 3
回答已采纳
我试图使用聚合功能的卡夫卡流翻滚窗口。但是，我得到了一个奇怪的构建错误。Grouped.with( MYBEAN_SERDE)) );org.apache.kafka
浏览 5提问于2022-03-21得票数 0
我有一个Kafka Streams应用程序(V2.1.1)，它产生的记录，并放入键值格式的输出主题。 key是窗口时间serde，其中我期望到窗口开始/结束时间的key和句柄。Windowed<String>> windowedSerde = Serdes.serdeFrom(windowedSerializer, windowedDeSerializer); 还有一个叫做kafkakafka属性： kafkaConsumerProperties.put(ConsumerConfig.KEY_DESERIALIZ
浏览 63提问于2019-05-09得票数 1
回答已采纳
我在kafka流中有以下逻辑流： .groupByKey()
.count(TimeWindows.of(3600
浏览 4提问于2017-10-15得票数 0
回答已采纳
2回答
来自火花流的背景-掌握卡夫卡流。并在该分钟内返回每个用户的最新事件。:30.754Z", "count": 3}, {"user": 1, "timestamp": "2018-05-22T16:56:39.754Z", "count": 4}
我感兴趣的是这将如何在Kafka流中工作，因为似乎每个事件都有一个输出--当我的用例是
浏览 0提问于2018-05-22得票数 1
回答已采纳
我正在尝试对具有字符串类型的键和自定义类型的值的流执行聚合，如下所示 stream.groupByKey(Grouped.with(Serdes.String(),barSerde)) .aggregate(Foo::new,
(String函数中的状态存储之前，我一直能够获得所需的结果，如下所示 stream.groupByKey(Grouped.with(Serdes.String(),barS
浏览 20提问于2019-06-12得票数 0
回答已采纳
云服务器
ICP备案
对象存储
云直播
腾讯会议
腾讯云开发者
