了解Kafka流组by和window

Question

我无法理解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一次打印一次，即先打印然后退出

Answer 1

似乎您的输入数据中没有键（如果这是错误的，请更正我），并且进一步看来，您想进行全局聚合吗？

通常，分组是用于将流分成子流。这些子流是按密钥构建的（即每个密钥一个逻辑子流）。您将时间戳记设置为代码段中的键，从而每个时间戳记生成一个子流。我认为这不是故意的。

如果要进行全局聚合，则需要将所有记录映射到单个子流，即，将相同的键分配给中的所有记录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/