Kafka流-更新KTable上的聚合

Question

我有一个KTable，数据如下所示(key=>value)，其中keys是客户ID，而value是包含一些客户数据的小型JSON对象：

1 => { "name" : "John", "age_group":  "25-30"}
2 => { "name" : "Alice", "age_group": "18-24"}
3 => { "name" : "Susie", "age_group": "18-24" }
4 => { "name" : "Jerry", "age_group": "18-24" }

我想对这个KTable做一些聚合，基本上保留每个age_group的记录数。所需的KTable数据如下所示：

"18-24" => 3
"25-30" => 1

假设Alice属于上面的18-24组，她的生日使她进入了新的年龄组。支持第一个KTable的状态存储现在应该如下所示：

1 => { "name" : "John", "age_group":  "25-30"}
2 => { "name" : "Alice", "age_group": "25-30"} # Happy Cake Day
3 => { "name" : "Susie", "age_group": "18-24" }
4 => { "name" : "Jerry", "age_group": "18-24" }

我希望得到的聚合KTable结果反映这一点。例如。

"18-24" => 2
"25-30" => 2

我可能过度概括了这里所描述的问题：

在Kafka的溪流中，没有最终聚合这回事。根据您的用例，手动去重复将是解决问题的一种方法。“

"18-24" => 3 # Old Alice record still gets counted here
"25-30" => 2 # New Alice record gets counted here as well

我使用的拓扑看起来像：

dataKTable = builder.table("compacted-topic-1", "users-json")
    .groupBy((key, value) -> KeyValue.pair(getAgeRange(value), key))
    .count("age-range-counts")

现在，从最初的，空的状态来看，一切都是这样的：

compacted-topic-1
(empty)


dataKTable
(empty)


// groupBy()
Repartition topic: $APP_ID-age-range-counts-repartition
(empty)

// count()
age-range-counts state store
(empty)

现在，让我们向compacted-topic-1发送一条消息，该消息在上面以KTable的形式流化。以下是发生的情况：

compacted-topic-1
3 => { "name" : "Susie", "age_group": "18-24" }
4 => { "name" : "Jerry", "age_group": "18-24" }

dataKTable
3 => { "name" : "Susie", "age_group": "18-24" }
4 => { "name" : "Jerry", "age_group": "18-24" }


// groupBy()
// why does this generate 4 events???
Repartition topic: $APP_ID-age-range-counts-repartition
18-24 => 3
18-24 => 3
18-24 => 4
18-24 => 4

// count()
age-range-counts state store
18-24 => 0

null

Answer 1

如果您的原始ktable包含id->json数据（我们称之为dataktable)，那么您应该能够通过以下方式获得所需的内容

KTable countKTablePerRange
    = dataKTable.groupBy(/* map your age-range to be the key*/)
                .count("someStoreName");

这应该适用于Kafka的Streams API的所有版本。

更新