如何使用 JAVA 对 kafka 流中的 AVRO 格式数据执行连接操作答案

【问题标题】：How to perform join operation on AVRO formatted data in kafka streams using JAVA如何使用 JAVA 对 kafka 流中的 AVRO 格式数据执行连接操作
【发布时间】：2018-10-31 00:04:57
【问题描述】：

STREAM-1：

 [KSTREAM-SOURCE-0000000000]: null, {"id": 1, "name": "john", "age": 26, "updated_at": 1525774480752}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 2, "name": "jane", "age": 24, "updated_at": 1525774480784}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 3, "name": "julia", "age": 25, "updated_at": 1525774480827}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 4, "name": "jamie", "age": 22, "updated_at": 1525774480875}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 5, "name": "jenny", "age": 27, "updated_at": 1525774482927}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 6, "name": "kishore", "age": 27, "updated_at": 1525775063908}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 7, "name": "purna", "age": 27, "updated_at": 1525775072006}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 8, "name": "xxx", "age": 10, "updated_at": 1525783464123}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 9, "name": "yyy", "age": 10, "updated_at": 1525783667644}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 10, "name": "zzz", "age": 10, "updated_at": 1525783741814}

STREAM-2：

[KSTREAM-SOURCE-0000000002]: null, {"id": 1, "name": "d", "age": 67}
[KSTREAM-SOURCE-0000000002]: null, {"id": 2, "name": "e", "age": 78}
[KSTREAM-SOURCE-0000000002]: null, {"id": 12, "name": "d", "age": 67}
[KSTREAM-SOURCE-0000000002]: null, {"id": 21, "name": "e", "age": 78}

现在我想对两个流执行 JOIN 操作，并且只想检索流 1 中不存在于流 2 中的行。我的输入流数据是 AVRO 格式

预期输出：

 [KSTREAM-SOURCE-0000000000]: null, {"id": 3, "name": "julia", "age": 25, "updated_at": 1525774480827}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 4, "name": "jamie", "age": 22, "updated_at": 1525774480875}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 5, "name": "jenny", "age": 27, "updated_at": 1525774482927}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 6, "name": "kishore", "age": 27, "updated_at": 1525775063908}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 7, "name": "purna", "age": 27, "updated_at": 1525775072006}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 8, "name": "xxx", "age": 10, "updated_at": 1525783464123}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 9, "name": "yyy", "age": 10, "updated_at": 1525783667644}
 [KSTREAM-SOURCE-0000000000]: null, {"id": 10, "name": "zzz", "age": 10, "updated_at": 1525783741814}

那么我应该执行哪个 JOIN 操作以及如何实现我的预期输出？谁能帮我实现这一目标

【问题讨论】：

标签： apache-kafka avro kafka-consumer-api apache-kafka-streams

【解决方案1】：

如果您查看此处的文档：kafka streams join semantics，您可能会使用 Left Join 并在设置 stream2 的值时在值连接器中返回 null。

一些伪代码：

stream1.leftJoin(stream2, valueJoiner);

valueJoiner = (s1, s2) -> {if (s2 != null) {
      return false
   } else {
     return true;
   } 
};

免责声明：我没有对此进行测试。

【讨论】：

感谢您的回复。这是在 JAVA 7 中知道的，但是，你能告诉我如何在 JAVA 8 中编写它，因为我是这个 AVRO 格式数据的新手。 @丹尼尔菲格罗亚