Как выполнить операцию соединения с данными в формате AVRO в потоках kafka с использованием JAVA

Question

Как выполнить операцию соединения с данными в формате AVRO в потоках kafka с использованием JAVA

ПОТОК-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}

ПОТОК-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 я должен выполнить и как добиться ожидаемого результата? Может ли кто-нибудь помочь мне добиться этого

0

kafka-consumer-api apache-kafka-streams avro confluent confluent-kafka

Источник

user6399990 21 май '18 в 10:38

1 ответ

Другие вопросы по тегам kafka-consumer-api apache-kafka-streams avro confluent confluent-kafka

user520684 21 май '18 в 12:23 2018-05-21 12:23 · Answer 1 · 2018-05-21 12:23

Если вы посмотрите документацию здесь: семантика объединения потоков kafka, вы, возможно, могли бы использовать левое соединение и просто возвращать нуль в вашем соединителе значений, когда установлены значения из stream2.

Некоторый псевдокод:

stream1.leftJoin(stream2, valueJoiner);

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

Отказ от ответственности: я не проверял это.