search

enybady / ks_test

0 stars 0 forks source link

Реализовать kafka-streams приложение #1

Open enybady opened 5 years ago

enybady commented 5 years ago

DoD:

dnoskov commented 5 years ago

Источники:

Из источников строго обязательный — документация kafka. По scala помимо документации есть книга: Кей Хорстман "Scala для нетерпеливых" — для быстрого введения. Из sbt нужно знать в основном команду: sbt run — она запускает основной класс проекта из текущей директории. Также может оказаться полезной команда sbt console — запускает REPL scala в контексте текущего проекта — полезно для экспериментов.

eugene-eliseev commented 5 years ago

Есть пара вопросов

  1. По каким причинам приложение Kafka Streams/приложение Kafka Consumer может не читать входной топик? Соединение с Kafka есть - можно просмотреть список всех топиков. При этом через консоль (kafka-console-consumer/producer) всё работает. Может какие-то нюансы есть? Код могу залить.
  2. Сейчас у меня немного костыльное решение с созданием дополнительного топика. Не факт, что оно вообще работает (не могу проверить из-за п.1), но просто надо было написать хоть какую-то основу для дальнейшего тестирования. Годится ли такое решение или лучше обойтись без создания доп. топика? (У меня есть ещё пара вариантов)
eugene-eliseev commented 5 years ago

Поднимал Kafka через докер из следующих образов: wurstmeister/zookeeper wurstmeister/kafka:latest

eugene-eliseev commented 5 years ago

Кажется, что-то с сетевым доступом не так. Запустил Producer на самой виртуалке с Kafka - всё заработало

eugene-eliseev commented 5 years ago

В общем, что не так было с сетью - позже разберусь, 1й вопрос отменяется

eugene-eliseev commented 5 years ago

На 2й вопрос, кажется, тоже нашёл ответ

eugene-eliseev commented 5 years ago

Вопрос по заданию - окно, в котором мы считаем напечатанные слова, должно быть скользящее? Если так, какой период сдвига у него должен быть?

dnoskov commented 5 years ago

@eugene-eliseev в общем любое окно, работу которого можно проверить за вменяемое время. Скользящее (sliding) окно в kafka-streams опирается на временные метки и не имеет параметра-прыжка. Для hopping window пусть будет прыжок - 1 минута

eugene-eliseev commented 5 years ago

У меня вроде получилось сделать, но есть один нюанс. Для теста я использовал окно 10 секунд и прыжок 2 секунды. Так вот данные в топик выкидываются не раз в 2 секунды, а сразу кучей раз в 30 секунд (то есть данные собираются каждые 2 секунды, скапливаются и выкидываются каждые 30 секунд). Это нормально или я что-то делаю неправильно (не указал чего-то)?

eugene-eliseev commented 5 years ago

Ну и соответственно второй топик, который основывается на предыдущем, запаздывает, как раз на 30 секунд

eugene-eliseev commented 5 years ago

Вопрос не актуален, забыл дописать кое-что )

eugene-eliseev commented 5 years ago

Закоммитил скрипт. Костыльный наверное, но пока какой есть ) Вроде работает )

enybady commented 5 years ago

пока только почитал, сразу несколько замечаний:

  1. В мастер все-таки не принято пушить, лучше в одтдельную ветку и пул реквест делать
  2. Не уверен что логично при редьюсе делать выбор в пользу первого элемента а не второго (нового)
  3. Попробуй вторую часть реализовать через аггрегацию таблицы а не приводить ее обратно к стриму
  4. Раз это пока код ревью то замечу, что есть throurh(). Найди что это и примени, будет полезно.
  5. Перед группировкой ты делаешь map, зачем?
  6. groupBy((key, _) => key) == groupByKey
  7. Зачем таскать сконкатенированную строку, если цель все равно разобрать ее в сет, почему с самого начала не таскать сет, просто испозовать не reduce а aggregate.

Запускать, окончательно проверять будем завтра

eugene-eliseev commented 5 years ago
  1. Понял, принял ) Я отталкивался от того, что для задания выделен отдельный репозиторий и чтобы не усложнять себе жизнь, дополнительных веток делать не стал
  2. Пожалуй да, поправлю
  3. Попробую
  4. Окей
  5. Ну я делаю ключ - первая буква, и потом по ней группирую
  6. Поменяю
  7. Я хотел с самого начала Set, но у меня потом ошибка возникала при группировке, так что для начала оставил такой костыль. Про aggregate собирался потом почитать, чтобы заменить все костыли на нормальный код