加入外键Kafka流

Question

假设我有三个Kafka主题，其中充满了表示不同聚合中发生的业务事件的事件（事件源应用程序）。这些事件允许构建具有以下属性的聚合：

• 用户：usedId，名称
• 应用程序的模块：moduleId，name
• 授予用户应用程序模块：grantId、userId、moduleId、作用域

现在，我想创建一个包含所有授权的流，其中包含用户和产品的名称（而不是id）。我想这么做：

1. 通过按用户ID对事件进行分组，为用户创建KTable。KTable将userId作为键。没关系

那么如何继续呢？

Answer 1

此功能作为Kafka Streams 2.4.0的一部分发布。

下面是关于使用此功能的官方教程。

Answer 2

嗯，目前在Kafka流中没有对外键连接的直接支持<有一个开放的基普：https://issues.apache.org/jira/browse/KAFKA-3705同样的。

目前，可以有一种变通方法来解决这个问题。您可以使用KStream KTable Join。

首先使用聚合的事件集合将用户流和模块流聚合到各自的KTable中。

KTable<String,Object> UserTable = userStream.groupBy(<UserId>).aggregate(<... build collection/latest event>) ;
KTable<String,Object> ModuleTable = moduleStream.groupBy(<ModuleId>).aggregate(<... build collection/latest event>);

现在选择moduleID作为授权流中的键。

KStream<String,Object> grantRekeyedStream = grantStream.selectKey(<moduleId>);

它将键更改为moduleId。现在您可以使用ModuleTable执行Stream-Table Join。它将从右侧为左侧的键加入所有匹配记录。结果流将以ModuleId为键将Grant和Module数据合并到一个流中。

KStream<String,Object> grantModuleStream = grantRekeyedStream.join(moduleTable);

下一步是加入userTable。因此，您需要再次使用userId重新键入grantModuleTable。

KStream<String,Object> grantModuleRekeyedStream = grantModuleTable.selectKey(<Select UserId>);

现在grantModuleRekeyedStream可以通过KStream-KTable Join与userTable连接

 KStream<String,Object> grantModuleUserStream = grantModuleRekeyedStream .join(userTable);

上述流将用户ID作为密钥，并包含该用户的所有授权和模块详细信息。