我无法理解 kafka 流中的 groupBy/groupById 和窗口的概念。我的目标是聚合一段时间(例如 5 秒(的流数据。我的流数据如下所示：
{"value":0,"time":1533875665509}
{"value":10,"time":1533875667511}
{"value":8,"time":1533875669512}
时间以毫秒(纪元(为单位。在这里，我的时间戳在我的消息中而不是在密钥中。我想平均 5 秒窗口的值。
这是我正在尝试的代码，但似乎我无法让它工作
builder.<String, String>stream("my_topic")
.map((key, val) -> { TimeVal tv = TimeVal.fromJson(val); return new KeyValue<Long, Double>(tv.time, tv.value);})
.groupByKey(Serialized.with(Serdes.Long(), Serdes.Double()))
.windowedBy(TimeWindows.of(5000))
.count()
.toStream()
.foreach((key, val) -> System.out.println(key + " " + val));
即使主题每两秒生成一次消息，此代码也不会打印任何内容。当我按 Ctrl+C 时，它会打印类似的东西
[1533877059029@1533877055000/1533877060000] 1
[1533877061031@1533877060000/1533877065000] 1
[1533877063034@1533877060000/1533877065000] 1
[1533877065035@1533877065000/1533877070000] 1
[1533877067039@1533877065000/1533877070000] 1
这个输出对我来说没有意义。
相关代码：
public class MessageTimeExtractor implements TimestampExtractor {
@Override
public long extract(ConsumerRecord<Object, Object> record, long previousTimestamp) {
String str = (String)record.value();
TimeVal tv = TimeVal.fromJson(str);
return tv.time;
}
}
public class TimeVal
{
final public long time;
final public double value;
public TimeVal(long tm, double val) {
this.time = tm;
this.value = val;
}
public static TimeVal fromJson(String val) {
Gson gson = new GsonBuilder().create();
TimeVal tv = gson.fromJson(val, TimeVal.class);
return tv;
}
}
问题：
为什么需要传递序列化程序/反序列化程序以分组依据。一些重载也会占用 ValueStore，那是什么？分组时，数据在分组流中的外观如何？
窗口流与组流有何关系？
以上，我期待以流媒体方式打印。这意味着每 5 秒缓冲一次，然后计数然后打印。它仅在命令提示符下按 Ctrl+c 后打印，即打印然后退出
您的输入数据中似乎没有键(如果这是错误的，请纠正我(，并且进一步似乎您想进行全局聚合？
通常，分组用于将流拆分为子流。这些子流是按键构建的(即，每个键一个逻辑子流(。您将时间戳设置为代码片段中的键，从而为每个时间戳生成子流。我认为这不是故意的。
如果要进行全局聚合，则需要将所有记录映射到单个子流，即为groupBy()中的所有记录分配相同的键。请注意，全局聚合不会缩放，因为聚合必须由单个线程计算。因此，这仅适用于小型工作负载。
窗口化应用于每个生成的子流以生成窗口，并且聚合按窗口计算。这些窗口是基于Timestamp提取器返回的时间戳构建的。似乎您已经有一个实现可以为此目的提取值的时间戳。
即使主题每两秒生成一次消息，此代码也不会打印任何内容。当我按 Ctrl+C 时，它会打印类似的东西
默认情况下，Kafka Streams 使用一些内部缓存，缓存将在提交时刷新 - 默认情况下每 30 秒刷新一次，或者在您停止应用程序时发生一次。您需要禁用缓存才能更早地看到结果(参见 https://docs.confluent.io/current/streams/developer-guide/memory-mgmt.html(
为什么需要传递序列化程序/反序列化程序以分组依据。
因为数据需要重新分发，而这是通过 Kafka 中的一个主题发生的。请注意，Kafka Streams 是为分布式设置而构建的，同一应用程序的多个实例并行运行以横向扩展。
顺便说一句：在这篇关于Kafka Streams执行模型的博客文章中，我们也可能很有趣：https://www.confluent.io/blog/watermarks-tables-event-time-dataflow-model/
您似乎误解了窗口DSL的本质。
它适用于 kafka 平台处理的内部消息时间戳，不适用于特定消息类型中编码时间信息的任意属性。此外，此窗口不分组为间隔 - 它是一个滑动窗口。这意味着您获得的任何聚合都是当前消息之前的最后 5 秒。
此外，您需要相同的键才能将所有组元素合并到同一组中，例如null。在您的示例中，key是一个条目唯一的时间戳，因此组中只有一个元素。