Kafka Streams 时间窗口聚合答案

【问题标题】：Kafka Streams Time Windowed AggregationKafka Streams 时间窗口聚合
【发布时间】：2020-12-21 15:27:59
【问题描述】：

我有一个流应用程序，具有以下域：

case class Test (testid: String, testcount: Long, testUser: String)
case class TestAgg (testid: String, aggCount: Long)

我有一个testStream，它从输入主题（testid 是关键）读取到 Test 案例类。这工作得很好。

但是，当我尝试在 testStream 上进行窗口聚合时，我观察到了奇怪的行为：

val aggStore = 
    testStream
    .mapValues((testid, test) => TestAgg(test.testid, test.testCount))
    .groupByKey
    .windowedBy(TimeWindows.of(Duration.ofMinutes(1)))
    .aggregate[TestAgg](TestAgg("", 0L))(
      (testid: String, n: TestAgg, o: TestAgg) => {
        TestAgg(n.testid, n.aggCount + o.aggCount)
      }
    ) (Materialized.as[String, TestAgg, ByteArrayWindowStore])("Test-agg-store").withKeySerde(Serdes.String()).withValueSerde(TestAgg.TestAggBytesSerDe)


aggStore
  .toStream
  .map((k,v) => (k.key(), v.asJson.toString()))
  .print(sysout)

我在控制台中看到以下输出：

[console]: Test_89, {
  "testid" : "Test_89",
  "aggCount" : 60515984
}
[console]: Test_33, {
  "testid" : "Test_33",
  "aggCount" : 45388033
}
[console]: Test_48, {
  "testid" : "Test_48",
  "aggCount" : 15130551
}

但是，我每 10 秒看到一次输出打印到控制台。它不应该每 1 分钟打印一次吗？我在这里错过了什么？

如果我添加，则取消：

aggStore
  .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
  .toStream
  .map((k,v) => (k.key(), v.asJson.toString()))
  .print(sysout)

根本没有输出。

我该如何解释这种行为？

编辑：

我换了窗口：

 .windowedBy(TimeWindows.of(Duration.ofMinutes(1L))
  .grace(Duration.ofSeconds(1L)))

之后我观察到两件事：

无抑制：流每 10 秒发出一次。我不知道为什么要 10 秒
带抑制：

.suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()).withName("Test-suppress-store"))

流确实每 1 分钟发出一次。（所以添加grace 一定有帮助）

但我最初关于观察 1 的问题仍然存在：为什么是 10 秒，为什么不是 1 分钟？我没有在任何地方指定 10 秒。

【问题讨论】：

标签： scala apache-kafka apache-kafka-streams

【解决方案1】：

您有正在处理的记录的时间戳吗？如果没有抑制，即使窗口没有关闭，每次处理新记录时都会发出窗口。

【讨论】：