Kafka流窗口是如何工作的？

Question

我很难理解窗口在Kafka Streams中是如何工作的。到目前为止，结果似乎与我所阅读和理解的不一致。

我已经创建了一个带有支持主题的KSQL流。KSQL SELECT语句中的“列”之一已被指定为该主题的TIMESTAMP。

CREATE STREAM my_stream WITH (KAFKA_topic='my-stream-topic', VALUE_FORMAT='json', TIMESTAMP='recorded_timestamp') AS select <select list> PARTITION BY PARTITION_KEY;

my-stream主题中的记录按键（PARTITION_KEY）分组，并用跳转窗口窗口

val dataWindowed: TimeWindowedKStream[String, TopicValue] = builder.stream('my-stream-topic', consumed) 
    .groupByKey(Serialized.`with`(Serdes.String(), valueSerde))
    .windowedBy(TimeWindows.`of`(TimeUnit.MINUTES.toMillis(5)).advanceBy(TimeUnit.MINUTES.toMillis(1)).until(TimeUnit.MINUTES.toMillis(5)))

记录通过

val dataAgg: KTable[Windowed[String], ValueStats] = dataWindowed
    .aggregate(
      new Initializer[TopicStats] {<code omitted>}},
      new Aggregator[String, TopicValue, TopicStats] {<code omitted>}},
      Materialized.`as`[String, TopicStats, WindowStore[Bytes, Array[Byte]]]("time-windowed-aggregated-stream-store")
        .withValueSerde(new JSONSerde[TopicStats])
    )

  val topicStats: KStream[String, TopicValueStats] = dataAgg
    .toStream()
    .map( <code omitted for brevity>)

然后我通过

dataAgg.print()
topicStats.print()

组中的第一个窗口转换为7:00-7:05

当我通过控制台消费者检查流主题中的记录时，我看到有两条记录应该在上面的窗口中。然而，聚合器只提取其中的一个。

我认为dataAgg窗口KTable将包含1条分组键的记录，但聚合将使用2条记录来计算聚合。打印的聚合值不正确。

我错过了什么？

Answer 1

KSQL可以在写入时设置记录时间戳，但是您需要在创建输入流时指定时间戳，而不是在定义输出流时。即，为输入流指定的时间戳将用于在写入时设置记录元数据字段。

这种行为相当不直观，我为此问题开了一张罚单：https://github.com/confluentinc/ksql/issues/1367

因此，在为问题中显示的查询创建输入流时，您需要指定with（TIMESTAMP='recorded_timestamp'）子句。如果这是不可能的，因为您的查询需要在不同的时间戳上运行，您需要指定将数据复制到新主题中的第二个查询。

CREATE STREAM my_stream_with_ts
    WITH (KAFKA_topic='my-stream-topic-with-ts')
AS select * from my_stream PARTITION BY PARTITION_KEY;

作为替代方案，您可以为您的Kafka Streams应用程序设置自定义时间戳提取器，以从有效负载中提取时间戳。