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

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

0 голосов
/ 21 мая 2018

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

1 Ответ

0 голосов
/ 21 мая 2018

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

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

stream1.leftJoin(stream2, valueJoiner);

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

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

...