KStream 通过非键值加入 GlobalKTable

Question

我正在尝试将 KStream 与 GlobalKTable 连接起来，连接不完全基于键。

GlobalKTable<String, Employee> employeesDetails = builder.globalTable("EMPLOYEE_TOPIC",..);
KStream<String,String> empIdOverLoginUserId = builder.stream("LOG_TOPIC", ….);

我想通过 empIdOverLoginUserId 的值通过 employeesDetails 的键加入 empIdOverLoginUserId 和 employeesDetails。

有线索吗？

Answer 1

KStream-GlobalKTable连接的第二个参数是一个KeyValueMapper映射KStream记录（key-value）到你要连接的GlobalKTable的key，你可以使用这个在连接时使用empIdOverLoginUserId的值作为key与 GlobalKTable:

empIdOverLoginUserId.join(
                employeesDetails, 
                (userKey, userValue) -> userValue, 
                (userValue, employeesDetailValue) -> employeesDetailValue)

KStream 通过非键值加入 GlobalKTable

KStream Join with GlobalKTable over non-key values

apache-kafka

apache-kafka-streams