Почему стриминг на KSQL и Kafka Streams — это непросто

[jonSina]

Почему при stream+table когда приходит со стороны table нету тригера? что это за бред?

Например у нас полностью асинхронный сайт:
1) добавляются видео (генерим guid, отправляем в очередь)
2) в UI можно сделать сразу допустим like (отправляем в очередь) (не придирайтесь я для простаты восприятия).

окей для видео TABLE — всё логично.
НО специально делать table для лайков вместо stream, из за того что не можем гарантировать кто первый придет, это бред!

[Lonsdaleite]

Во-первых, не особо понятно, как при stream-table join должен работать триггер справа. В стриме могут быть дубли по ключу, на что джойниться?
Во-вторых, Kafka Streams API писали не мы, возмущаться тут бессмысленно.

[jonSina]

учите документацию, вы пишите статью о граблях, не зная о таком лимите…

[Lonsdaleite]

Не понял, чего и о каком лимите, по вашему, я не знаю? Про то, что stream-table join триггерит джойн только приходом записей с левой части — это очевидный факт, так работает KSQL с Kafka Streams API

[bex2014]

пока похоже просто на рекламный пост… почему именно так сделали как на схеме? зачем там например «золотой по деньгам» golden gate оракловый? Можно про это именно рассказать?

[Lonsdaleite]

Лицензия на Golden Gate уже была. К тому же, это довольно хорошо работает. Из альтернативы, для загрузки данных из Оракла видится только какой-нибудь JDBC-коннектор, который работает ощутимо хуже и медленнее. GoldenGate льет данные в режиме реального времени, что является явным преимуществом.

[easty]

Не нашел примерные нагрузки. Не поделитесь что летает на такой конфигурации ПО и хардваре, приемные цифры потоков?

[Lonsdaleite]

Точных цифр по памяти сейчас не назову, надо лезть, смотреть и считать.
В целом, вопрос нагрузки интересный, вас что именно интересует? Если смотреть на голые цифры количества событий на вход и на выход, будет не очень много, но это ни о чем не скажет. Внутри происходят сложные вычисления в режиме реального времени с кучей джойнов и агрегатов, что создаёт почти всю нагрузку на кластер.

[Alexei_987]

Сильно поддержку вопрос касательно нагрузки. О каких вообще цифрах входящих и выходящих событий идет речь? Хотя бы порядок чисел.

[igor_dumchykov]

На практике столкнулся еще с одними граблями: на основе стримов реализовать ретраи (перекинуть сообщение обратно в очередь если оно было запроцессено с ошибками) с конфигурируемым временем задержки. Штатного решения я не нашел, Thread.sleep() просто усыпил поток стримов (а потоков согласно архитектуре kafka streams выделяется пропорционально кол-ву партиций). Остановился на таком решении: закидывать все сообщения для ретраев в State Store и периодически скедулером доставать все сообщения и проверять сколько времени они там лежат, если это время >= сконфигурированного времени ретрая — перекидываем в очередь. Возможно есть более простое решение?

[Lonsdaleite]

Мы реализовывали нечто подобное. Схема примерно такая: делаем перекладку сообщений в боковой топик с задержкой N секунд от таймстемпа сообщения, потом перекладываем сообщения в исходный топик. Понятное дело, накладываем по пути нужные условия, чтобы переподкидывать не всё. Если не очень понятно, пишите в личку, объясню подробнее.

[Kordamon]

Спасибо за статью. Скажите, какая в итоге получается нагрузка «с другой стороны трубы» — на RTDM? Желательно в TPS и MBPS

[Lonsdaleite]

В понедельник смогу посмотреть, скорее всего. На вскидку там порядка нескольких тысяч сообщений в секунду. В Mb надо измерять.

[ipodman]

Привет, спасибо за статью.
С проблемой RocksDb на HDD мы тоже столкнулись и в итоге перенесли хранение стора в tmpfs. Подскажите как вы реализуете деплой без потери данных? По сути когда меняешь структуру стрима, то меняется структура хранения в сторе и новая версия приложения с ней уже не совоместима? Или с ksql таких проблем нет?

[Lonsdaleite]

Мы тоже используем tmpfs. С ksql ровно те же проблемы, что и с kafka streams. Действия от изменения какого-либо шага со state store зависят от ситуации. Если меняется топик, на который идет джойн как на table, достаточно очистить state store и changelog топик от этого шага, после чего вычитать топик-таблицу заново. Если меняется структура агрегатов, приходится делать частичный репроцессинг данных.