Kafka流-旧状态聚合

Question

我有一个KStream,其中包含从主题到1的数据,如下所示：

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

和KTable,构造如下：

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

..

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

稍后,主题To2中出现以下消息：

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

现在，我希望我的KTable能够反映这些变化，并且看起来像这样：

T1 -> { ["B2"] }

但看起来是这样的：

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

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

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

}

其中，set aggrege 在第一次调用（通过初始值设定项创建）时为[]，但在第二次调用时为[“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 value) 。那么您根本就不需要聚合。为了实现这一点，我建议您转换输入集

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

进入

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

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

如果您为当前的flatmap实现提供了代码，很乐意详细阐述更多内容。