Kafka Streams - 与旧状态的聚合

Question

我有一个 KStream，其中包含来自主题 to1 的数据，如下所示：

T1-KEY -> {T1}
T2-KEY -> {T2}

还有一个KTable，构造如下：

我正在使用 org.apache.kafka.streams.StreamsBuilder 从某个主题 to2 创建 KTable，如下所示：

A1-KEY -> { "A1", "Set": [
                          {"B1", "Rel": "T1"},
                          {"B2", "Rel": "T1"}
                         ]
          } 

..

该流随后被平面映射并按 Key s.t. 分组。生成的 KTable 如下所示：

T1 -> { ["B1", "B2"] }

稍后，现在主题 to2 中出现以下消息：

A1-KEY -> { "A1", "Set": [
                          {"B2", "Rel": "T1"}
                         ]
          } 

现在我希望我的 KTable 能够反映这些变化并如下所示：

T1 -> { ["B2"] }

但它看起来像这样：

T1 -> { ["B1", "B2"] }

我注意到，在我的Aggregator<Tx-KEY, Bx, Set<Bx>> 中给出的最后一个参数是集合["B1", "B2"]，即使当我在聚合之前偷看时我只得到一个匹配"B2"。

我是否理解聚合错误或这里发生了什么？

编辑

我想我缩小了范围：显然聚合的 Initializer 只在 very 第一次调用 - 之后聚合总是接收 last aggregate 作为最后一个参数，例如

@Override
public Set<Bx> apply(Tx-KEY, Bx value, Set<Bx> aggregate) {

}

Set<Bx> aggregate 在第一次调用时是 []（通过初始化程序创建），而在第二次调用时是 ["B1", "B2"]。

有什么想法吗？

编辑 2

public class MyAggregator implements Aggregator<Tx-KEY, Bx, Set<Bx>> {

    @Override
    public Set<Bx> apply(Tx-KEY key, Bx value, Set<Bx> aggregate) {
        aggregate.add(value);
        return aggregate;
    }
}

编辑 3

我不能只使用平面地图，因为我必须组合多个 Ax 元素，例如

A1-KEY -> { "A1", "Set": [
                      {"B1", "Rel": "T1"}
                     ]
          },
A2-KEY -> { "A2", "Set": [
                      {"B2", "Rel": "T1"}
                     ]
          },
...

然后我期待一些类似的分组

T1 -> { ["B1", "B2"] }

并且在下一次迭代中，当消息

A1-KEY -> { "A1", "Set": [
                      {"B1", "Rel": "T1"}
                     ]
          }

如期而至

T1 -> { ["B1"] }

..

Answer 1

请注意，在您的聚合器中，您只会将元素添加到聚合集。使用这种逻辑，您的集合（对于给定的键）永远不会缩小。在这种情况下，我认为您将流压扁了太多。我建议您不要将其展平到您的消息采用(Tx-KEY key, Bx value) 的形式，而是让它们始终保留其设置的形式：(Tx-KEY key, Set<Bx> value)。你根本不需要聚合。 为此，我建议您转换输入集

"Set": [
     {"B1", "Rel": "T1"},
     {"B2", "Rel": "T1"}
]

进入

T1 -> { ["B1", "B2"] }

通过在 KStream flatmap 方法调用中使用标准 java 代码（Collections 或 Streams api）按“Rel”字段分组，这样您就只会在 KStream 上发出带有Set<Bx> 类型值的消息，而不是Bx - 单独键入值。

如果您提供当前平面地图实现的代码，我们很乐意详细说明。