Что такое Kafka? Где стоит, а где не стоит применять этот инструмент? Чем Kafka отличается от RabbitMQ и других брокеров сообщений? Как её правильно эксплуатировать? Всё это обсудили на митапе Apache Kafka в вопросах и ответах, который Слёрм провёл в ноябре 2020. В разговоре участвовали спикеры из Авито, Stripe, ITSumma и Confluent. Запись митапа доступна на YouTube, а текстовую версию разговора читайте ниже.

Знакомство со спикерами

МАРСЕЛЬ ИБРАЕВ: Сегодня мы говорим о Kafka, но не про философию, а про приложение. У нас очень крутые спикеры, разрешите мне их представить и сразу задать каждому приветственный вопрос: какой RPS на вашем проекте?

Анатолий Солдатов lead engineer в Avito, много работал с базами данных, Kafka, Zookeeper, ClickHouse и много выступает на митапах и конференциях. Если не секрет, какой RPS на вашем проекте?

АНАТОЛИЙ СОЛДАТОВ: У нас около 600 RPS и порядка 25 Гб/с. Это в сумме все кластера.

МАРСЕЛЬ ИБРАЕВ: Огонь! Также с нами Александр Миронов, Infrastructure Engineer из компании Stripe, занимается развитием системы CI, работал в таких компаниях, как 2ГИС, Lingualeo и 4 года возглавлял инфраструктурную команду разработки сервисов стриминга данных в Booking.com. И это нам о чём-то да должно говорить: стриминг, большие данные, и вот это вот все наверняка как-то связано. Правильно я понимаю, Александр? И какой RPS?

АЛЕКСАНДР МИРОНОВ: Да, ты понимаешь все совершенно правильно, на середину 2020 года в Booking.com RPS получалось около 60 Гб входящего трафика и около 200 Гб исходящего, в миллионах сообщений я, честно говоря, не помню. Но это десятки миллионов сообщений в секунду.

МАРСЕЛЬ ИБРАЕВ: Супер.

АНАТОЛИЙ СОЛДАТОВ: Мы, кстати, у ребят из Booking.com учились. У нас Kafka помоложе, и Саша нам очень сильно помогал затащить всё это. Делились опытом.

МАРСЕЛЬ ИБРАЕВ: Далее у нас Виктор Гамов, Developer Advocate в Confluent. Специализируется на обработке потоковых данных с помощью Kafka. Также спикер российских, международных IT-конференций.

ВИКТОР ГАМОВ: И стример. Стал стримить, вот до чего жизнь довела! На конференции не пускают, приходится стримить. Всем привет! Какой RPS я не знаю, потому что я бездельник, маркетолог, я Kafka только продаю, не замеряю её вес. На самом деле, у нас свой Public Cloud, и у нас есть и большие, и маленькие клиенты. Не могу конкретно о цифрах говорить, но клаудный бизнес растет достаточно хорошо, люди приходят за managed-решениями. Есть SLA до трех девяток, не всегда дело в RPS, а еще иногда и в SLA, и в других вещах, которые связаны с reliability, не только со скоростью.

МАРСЕЛЬ ИБРАЕВ: Не RPS единым.

ВИКТОР ГАМОВ: Да.

МАРСЕЛЬ ИБРАЕВ: Ну и наконец Иван Сидоров, Chief Technology Innovation Officer из компании ITSumma. Самое интересное, что Иван прошел путь от простого разработчика к архитектору и обратно несколько раз, совместив это с менеджментом. Более того, вдохновившись книгой по Kafka, он инициировал создание технического издательства в компании. К Ивану вопрос такой же: как используете Kafka, какие есть максимально нагруженные проекты? Есть ли такие данные и можно ли их озвучить?

ИВАН СИДОРОВ: Отвечу немного уклончиво. По должности я исполнительный директор компании и занимаюсь внедрением различных инноваций, моя работа узнавать про новые технологии, применять их в своей деятельности. Компания у нас сервисная, поэтому своя Kafka нам если и нужна, то только какая-то карманная. Но мы поддерживаем около половины Рунета крупного точно, и Kafka мы видели очень много, в разных конфигурациях, в разных вариантах использования. И RPS тут совершенно неважен. Тут важно, как используется Kafka, для чего она нужна. Конкретные цифры сказать не могу, но очень большие бывают тоже.

МАРСЕЛЬ ИБРАЕВ: Теперь, я думаю, наша уважаемая аудитория понимает, что компетенции участников достаточно высоки. Определенно будет, о чем поговорить.

АНАТОЛИЙ СОЛДАТОВ: Чем больше RPS, тем выше компетенция. Саша выиграл здесь, мне кажется.

АЛЕКСАНДР МИРОНОВ: Не, ну, Public Cloud от Confluent, я думаю, выиграет в любом случае.

АНАТОЛИЙ СОЛДАТОВ: Наверное, да, выиграет.

МАРСЕЛЬ ИБРАЕВ: Сам придумал такую линейку, сам решил.

ВИКТОР ГАМОВ: Вот мы это и делаем. Появилась Kafka, давайте из нее сделаем Kafka для всего, вот. У нас теперь Kafka для всего.

Что такое Kafka

МАРСЕЛЬ ИБРАЕВ: Много раз звучало слово Kafka, и я бы хотел, чтобы наша аудитория вышла на один уровень понимания, что же такое Kafka и чем оно не является. В определениях часто встречается такое слово, как брокер сообщений. И для меня, человека рефлексирующего по 2007 году, это словосочетание вызывает ассоциации с Jabber, ICQ и прочим. Все-таки хочется понять, что же такое на самом деле Kafka и для чего она предназначена.

АНАТОЛИЙ СОЛДАТОВ: Давайте по вариантам, чтобы было интереснее. Это лог, это стриминговая платформа, это база данных, свой вариант?

ВИКТОР ГАМОВ: Свой вариант. В книжке от Бена Стоффеда есть замечательная метафора: слепые встретили слона и начали его ощупывать. И в зависимости от того, что они трогали, у всех складывалось определенное впечатление. Кто-то потрогал за хвост и подумал, что это опоссум. Кто-то потрогал за хобот и подумал, что это удав. Кто-то потрогал за кожу и вообще ничего не понял. Такая же история и с Kafka. Все зависит от того, откуда вы пришли, от вашего изначального опыта и того, как вы ее начали трогать. Толя дал варианты: лог, стриминговая платформа, база данных? Я скажу так это всё. Попробуйте переубедить меня.

С Kafka интересная история. Можно сказать, что пришли маркетологи и заставили нас думать, что Kafka это такая панацея от всего. На самом деле, Kafka появилась из одной простой идеи. Эта идея не нова, но её решили выкатить в первый ряд. Это идея лога, и она всегда присутствовала в любой системе, где есть персистентность, будь то база данных или система обмена сообщениями. Лог это простая идея: у тебя есть файлик, ты в него пишешь просто в конец и читаешь с начала. Эту идею имплементировали сами не раз руками. Поэтому, когда приходит слепой (из той метафоры), эксперт из мира очередей, он смотрит: ну, это же очередь! Мы же пишем в конец и читаем из начала. Это же очередь. Значит, месседжинг, правильно? Правильно.

Дальше приходит человек из мира баз данных и говорит: погодите, но мы же в Kafka данные размазываем по брокерам. Там есть партицирование, там есть consistent hashing, когда мы по какому-то ключу вынимаем, по значению знаем, куда класть. Когда клиенту не надо иметь какого-то внешнего load balancer, чтобы договориться. Это же распределенная база данных, это же NoSQL во все поля. Плюс, потому что там есть persistence, это точно база данных.

Потом приходят люди из мира ESB (enterprise service bus) и говорят: ну, там же есть коннекторы, которые позволяют нам перетаскивать данные отсюда сюда. Это ж ESB! Есть небольшой коннектор, на коннекторе есть какие-то определенные SMT (single message transforms), которые позволяют раутить в минимальном объеме. Точно ESB!

Получается неразбериха. Поэтому решили остановиться на таком определении, как Event Streaming Platform платформа для захвата и обработки стримов, потоков сообщений, потоков событий. Событие это какая-то иммутабельная запись, которая не меняется, и в идеале постоянно лежит в Kafka.

ИВАН СИДОРОВ: Дополню. Ты перечислил много применений, а если у нас замешаны датчики, то это уже IoT интернет вещей. Kafka тоже там! А вообще, это такая же категория, как операционная система. Что делает операционная система? Все что угодно. Kafka это система для работы с данными. Соответственно, в каком контексте её используют, такие возможности она и предоставляет.

АНАТОЛИЙ СОЛДАТОВ: Да, еще забыл вариант труба. И ещё вопрос: встречали ли вы сетапы, где Kafka используется как база данных самостоятельно? Без всяких PostgreSQL, MongoDB и так далее. Вот есть приложение, есть Kafka, и больше ничего нет.

ИВАН СИДОРОВ: Мы да.

АНАТОЛИЙ СОЛДАТОВ: То есть, такие кейсы вполне себе есть?

АЛЕКСАНДР МИРОНОВ: Да, встречали.

АНАТОЛИЙ СОЛДАТОВ: А с ksqlDB что-то такое уже пробовали, чтобы это была полноценная замена реальной базе данных: с индексами, с селективными запросами, которые могут ходить в прошлое?

ИВАН СИДОРОВ: Кейсы, которые видели мы, были чем-то вроде хранилища логов. Сложные запросы не нужны были.

ВИКТОР ГАМОВ: В чате пишут, что я еще забыл самый правильный юзкейс, который Kafka использует RPS строить поверх нее.

ИВАН СИДОРОВ: А чем он отличается от ESB, например?

МАРСЕЛЬ ИБРАЕВ: Все зависли над твоим вопросом.

ВИКТОР ГАМОВ: Накручиваешь, накручиваешь, и получается сборка того, что тебе нужно. Года три назад была статья от The New York Times о том, как они используют Kafka в качестве хранилища. Но у них такое, каноническое хранилище источник информации, из которого все забирают и потом каким-то образом модифицируют. Тут нужна СMS, она все это вычитает и в какие-то форматы обернёт, чтобы отрисовать это всё на экране. Если это система каталогизации, она Kafka берёт как исходник, а поверх строит какой-то индекс.

В чате пишут, что Kafka это винегрет. Kafka это Kafka.

АЛЕКСАНДР МИРОНОВ: Такое количество способов использования системы, в очередной раз подчеркивает, что перед тем как начинать пользоваться Kafka, вам нужно чётко понимать, что именно вы от неё ожидаете, какую пользу для бизнеса вы планируете извлечь. Будь то open source Kafka или managed-решения вроде Confluent, и так далее.

Может быть, вы хотите обрабатывать данные и получать репорты для ваших аналитиков в реальном времени, а не через сутки это один способ использования. Или, может быть, как в Times, вы хотите сохранить всю историю изменений вашей базы данных. Возможно ли это для вашего юзкейс, разумеется, невозможно заранее ответить, потому что у Times, все топики занимали несколько терабайт, по-моему, а если у вас петабайты данных, вам наверняка будет намного сложнее это сделать.

ВИКТОР ГАМОВ: Но все ещё возможно, есть способы.

АНАТОЛИЙ СОЛДАТОВ: Возможно-то всё, вопрос здравого смысла.

ВИКТОР ГАМОВ: Это самое главное. Спасибо, что есть Толик, который приходит и говорит правильные мысли. Вопрос не в RPS, а в здравом смысле.

МАРСЕЛЬ ИБРАЕВ: Получается, Kafka это действительно комбайн, который сочетает в себе различную функциональность. Это может быть обработчик сообщений (брокер), база данных

Где стоит, а где не стоит применять

МАРСЕЛЬ ИБРАЕВ: В каких областях Kafka можно применить? Я так понимаю, это анализ данных. Ещё я сталкивался с системой логирования мониторинга. Где помимо этих сфер можно применять Kafka?

АНАТОЛИЙ СОЛДАТОВ: В чатике было дополнение, что Kafka хорошо подходит для обмена сервисов, то есть notification туда отлично ложатся.

Давай я перечислю, для чего Kafka используется в Avito, а то фантазировать много можно. Для аналитики, для всяких кликстримов, для стриминга, для message broker, для того, чтобы вставлять различные буферы перед системами типа Elastic или ClickHouse. У нас это хорошо задружилось с трейсингом, например. Такие в основном паттерны: аналитика, message broker, буфер, база данных.

АЛЕКСАНДР МИРОНОВ: Fanout еще, такой классический кейc.

ИВАН СИДОРОВ: Аналитика в каком плане?

АНАТОЛИЙ СОЛДАТОВ: Самый простой кейс это когда на платформу входит весь входящий трафик кликстримовый, а нашим аналитикам весь трафик не нужен, потому что там очень много ботов. И мы прямо в Kafka берём и фильтруем всё это. У нас появляется отфильтрованный топик, в котором содержится только чистый трафик, и он идёт, например, дальше в ClickHouse.

АЛЕКСАНДР МИРОНОВ: Еще один классический кейс это security-анализ, анализ фишинга. Потому многие фреймворки для реал-тайм анализа из коробки позволяют использовать sliding window пятиминутное, которое автоматически за вас будет пересчитывать счетчики. Это очень удобно.

МАРСЕЛЬ ИБРАЕВ: Огонь! У каждого инструмента есть лучшие/худшие практики: молоток хорошо забивает гвозди, но замешивать им тесто не очень удобно, хотя и возможно. Есть ли области, где Kafka не ложится никак?

ВИКТОР ГАМОВ: High-Frequency-Trading (HFT). Здесь Kafka ни о чём, потому что, где диск привязан, там начинаются проблемы. Плюс, это распределённые системы. Там, где начинаются всякие сетевые вопросы, скорее всего, Kafka не подойдет. Если нужен простой месседжинг с простым раутингом (без внедрения тех вещей, о которых мы уже поговорили вкратце, KSQL, Kafka-streams, флипинг и т.д), здесь Kafka тоже не подходит. Потому что нужен просто message broker. Можно, конечно, использовать Kafka. Но потом люди приходят и говорят: вот, Kafka для нашего HFT не подходит. Ну, естественно не подходит. Для этого есть какой-нибудь Aeron или ZeroMQ, которые были создан для того, чтобы не бежать по распределёнке, а
локально.

АНАТОЛИЙ СОЛДАТОВ: Тут в целом, если тебе нужны и очереди, и супер low latency, и 100% гарантия доставки, и у тебя поток данных небольшой, то лучше не искать огромных инструментов типа Kafka...

ИВАН СИДОРОВ: а использовать хардварные решения.

АНАТОЛИЙ СОЛДАТОВ: Ну, я подводил к RabbitMQ, конечно, но можно и так.

ИВАН СИДОРОВ: Нет, это долго.

ВИКТОР ГАМОВ: Правильное, на самом деле, уточнение по поводу хардварных решений. Я раньше трудился в компании, которая делала небольшой распределенный кэш, назывался Hazelcast. И мы делали такую интеграцию интерпрайзную со всякими балалайками, чтобы делать кросс-датацентерную (как вы по-русски говорите?) Multi-Datacenter Replication.

АЛЕКСАНДР МИРОНОВ: Очень по-русски сказал.

ВИКТОР ГАМОВ: И мы как раз перед тем, как запилить адаптер для Kafka, мы запилили адаптер для Solace, который давал хардварное решение. Люди с большими деньгами покупали платы, которые давали супербыстрый latency. Там вопрос был именно передачи информации между океаном, между Чикагской биржей и Лондонской, поэтому Solace давал какие-то дикие цифры, потому что лежал поверх кастомной сети и кастомного железа. Поэтому шутки шутками, но если хочется скорости, всегда можно опуститься, как Ваня сказал, на уровень железки.

ИВАН СИДОРОВ: Если ты в HFT, у тебя уже есть деньги для того, чтобы сделать кастомную железку.

ВИКТОР ГАМОВ: Или деньги инвесторов.

ИВАН СИДОРОВ: Да, а если нет, то HFT не будет.

АЛЕКСАНДР МИРОНОВ: Ну, и еще одна холиварная тема это использование Kafka в качестве базы данных. Теоретически это возможно и даже практически мы видим какие-то кейсы, но чтобы сделать это правильно, потребуется значительное количество времени, и во многих случаях, возможно, имеет смысл продолжить спокойно использовать PostgreSQL или что-то еще, что позволяет делать простые запросы по любым ключам. Потому что Kafka все-таки да, имеет выборку по офсету ID, но даже эта операция значительно медленнее...

ВИКТОР ГАМОВ: Вот ты правильно говоришь, но для слушателей надо немного пояснить. Почему Kafka ассоциируется с известным докладом Мартина Клеппмана (Martin Kleppmann) Turning the database inside-out (если вы не видели, сходите посмотрите). Отличие Kafka от базы данных в том, что и там, и там есть лог и durable-хранилище, которое обеспечивает хранение данных на долгий срок, но в обычной базе данных для вас в API есть Query Engine, с которым вы взаимодействуете посредством какого-то DSL: SQL, в данном случае, если noSQL, там тоже какой-то свой, JSON, не знаю, что-нибудь в MongoDB. В Kafka всё немного иначе. Там storage, собственно, вот этот лог, и ваш Query language разнесены. Поэтому абсолютно непрактично использовать какой-то стандартный low level кафковский API для того, чтобы бегать по апсету. Это не нужно, неправильно и вообще вредно.

Обычно смысл Kafka в том, что вы можете насадить туда любое приложение, и оно становится вот этим Query Agent, и вы дальше накручиваете. Вы хотите иметь какой-то хитрый язык запроса? Вы можете это сделать. Речь идет о том, что стандарта нет такого, чтобы вот взять и сделать из Kafka k-value какой-то Storage. Вам нужно все равно что-то либо написать, например, взять Kafka Streams, написать там две строчки, и у вас получается из топика сделать материализованный View и отдать его через REST. Это достаточно просто сделать. Такой же процесс использует skim-registry. Вот мы сделали балалаечку, которая позволяет хранить схемы, данные. Skim-registry не нужно ничего, кроме Kafka. Она внутри Kafka хранит все ваши схемы. Она наружу отдаёт REST-интерфейс. Если вы пришли и сказали: вот у меня есть такой тип схемы, вот мой номер ID, отдай мне всю схему, всё это реализовано внутри Kafka. Kafka используется как хранилище.

Kafka Сonnect использует Kafka как базу данных. Там тыща параметров всяких, настроек лежит внутри Kafka. Когда вы стартует, Сonnect голый, он все настройки хранит внутри Kafka. То есть, здесь ничего не надо. Это к вопросу о том, что Марсель сказал про молоток. Можно молотком помешать тесто? Можно. Можно ли на Kafka сделать базу данных? Можно. Можно ли на ней сделать серьезную базу данных? Можно. Вопрос в уровне упоротости. Насколько глубоко вы готовы пойти.

ИВАН СИДОРОВ: Kafka и база данных это звучит дико из-за того, что не сходится с парадигмой обычной базы данных, где язык запросов и база данных неотделимы друг от друга. Но это достаточно просто показывается на примере того же самого Hadoop, который стал уже достаточно расхожей технологией. Что такое Hadoop? Обычно, когда говорят Hadoop, говорят HDFS. Это просто файловая система, а поверх нее можно накручивать схемы, разные движки запросов, Hive, HBase, все что угодно. И в Kafka, как я понимаю, просто пока вот этих Query Engine которые бы уже закрепились в индустрии, пока нет. И поэтому Kafka для базы данных
воспринимается так.

И если вопрос о том, где Kafka применяется неправильно, чуть переиначить и спросить: а где нерационально или слишком дорого применять Kafka это логирование. Можно делать логирование на Kafka и писать вокруг что-то или можно взять готовый ELK, который ты клик и установил. Зачем тут Kafka? Нерационально. Но если нужна система логирования, которая потом обрабатывает кучу микросервисов, отправляется через Сonnect в какой то-то data warehouse и так далее, то тут уже надо подумать, что ELK не принесёт пользы, а принесет
больше вреда, пока будешь ковыряться внутри Elastic.

МАРСЕЛЬ ИБРАЕВ: Супер, с этим вопросом разобрались. Теперь понятно, что за инструмент такой Kafka. Пойдем дальше.

Kafka vs RabbitMQ

МАРСЕЛЬ ИБРАЕВ: На круглый стол зарегистрировались около полутора тысяч человек, они задали порядка 200 вопросов, 90-95% которых были про RabbitMQ.

И первый вопрос: в чем отличие? Как я понял, Kafka это многофункциональный инструмент. RabbitMQ работает с очередями и для этого предназначен. Так ли это?

АНАТОЛИЙ СОЛДАТОВ: По большому счету, наверное, да. RabbitMQ планировался в первую очередь для очередей, и по-другому мы его не использовали. Kafka дизайнилась как бы с другой парадигмы, что ли. Вопервых, там очереди и топики это разная семантика, они не повторяют свои свойства. Во-вторых, у Kafka была изначально цель высокая пропускная способность, реально высокая. RabbitMQ, я и на своем примере знаю, и от разных компаний слышал, что 20-30 K RPS для RabbitMQ это предел. Как очередь он в принципе и не должен выдерживать больше. Наверняка, есть какой-нибудь мастер RabbitMQ, который его так соберёт, что это будет суперскоростной реактивный RabbitMQ, но это тоже вопрос: а зачем он такой, если есть технологии под этот кейс? А для Kafka перемалывать сотни, миллионы RPS это нормально. Она дизайнилась именно для этих целей. Но, с другой стороны, если нужен low latency и вам важно, чтобы одно сообщение быстренько долетело до консюмера, нигде не потерялось, то здесь RabbitMQ может и лучше подойти.

АЛЕКСАНДР МИРОНОВ: Тут нужно уточнить, что Kafka всё равно можно затюнить под low latency, чтобы у вас были миллисекундные задержки. Но для этого нужно постараться.

АНАТОЛИЙ СОЛДАТОВ: Надо потрудиться, да. И это будет, скорее всего, выделенный кластер. Не получится сделать один кластер и под high-throughput и под low latency.

АЛЕКСАНДР МИРОНОВ: Самое главное, действительно, в разной семантике топиков и очередей. Виктор уже назвал десяток технологий подобных очередей. В концепции очереди у вас есть возможность удалить сообщение или пометить его как прочитанное. Вы можете это сделать в любом порядке. Kafka это лог, которому всегда аппендятся записи. Из него ничего удалить нельзя. Соответственно, единственная метка того, где сейчас находится ваш консюмер это текущий upset, который всегда монотонно возрастает (monotonic increasing). Соответственно, вы не можете просто взять и без дополнительных инструментов пропустить какое-то сообщение, чтобы потом к нему вернуться и прочитать. Вам пришло 10 сообщений, вы должны дать Kafka ответ: вы их все обработали или нет.

АНАТОЛИЙ СОЛДАТОВ: Не могу удержаться и не сказать, что так тоже можно

АЛЕКСАНДР МИРОНОВ: Можно, да. У Uber есть большая статья: Dead Letter Queues with Apache Kafka. Про то, как они сделали Dead letter queue.

АНАТОЛИЙ СОЛДАТОВ: Кстати, в чем отличие того же Dead letter queue в RabbitMQ оно встроено в брокер, а в Kafka клиенты или что-то там над брокерами реализует эту всю логику.

АЛЕКСАНДР МИРОНОВ: Вам понадобится ещё дополнительный функционал, чтобы это сделать. И возможно, можно взять просто out of the box инструменты, а уж тем более, если вы находитесь где-то в Public Cloud, в Amazon или в Google, у вас уже есть очереди, которые доступны из коробки.

АНАТОЛИЙ СОЛДАТОВ: Опять же, различия есть. Они тоже из топиков вытекают, как Саша сказал, что есть offset consumer-groups, в RabbitMQ ты не особо почитаешь несколькими консюмерами из одной очереди. Это можно сделать, размножив очередь. А в Kafka ты читаешь один и тот же лог, и у тебя апсеты трекаются независимо для разных consumer-groups. Это вполне себе легко работает.

ВИКТОР ГАМОВ: Вот это, кстати, такой очень интересный момент, который не всегда все до конца понимают. Люди спрашивают: как мне сделать так, чтобы кто-то другой не прочитал мое сообщение? Или наоборот, чтобы смог. Я думаю, это опять все связано с тем, что Kafka не появилась из научного института, где все было сразу продумано, она появилась от сохи: люди решали собственную проблему. Вот эта категория имен, которая пришла из месседжинга, и вызывает определенные вопросы. Топик, брокер это же все месседжинг. И люди накладывают поверх этого какой-то свой предыдущий бэкграунд.

Соответственно, когда у тебя есть возможность репроцессить лог и возвращаться к каким-то вещам, например, ты получил данные и знаешь, что у твоего поставщика данных нет возможности их перечитать. Они у тебя уже лежат в системе, и ты их можешь просто препроцессить, может быть, применить алгоритм. Или ты просто пытаешься что-то новое из этих документов получить. Какую-то информацию. Это уникальная вещь, которая есть в Kafka, и зачастую о ней забывают.

Но, как только мы начинаем говорить про вещи, как Event Sourcing, QSRs, вот эти все микросервисы обмена сообщениями, вот тут люди начинают видеть бенефиты. Если они приходят в Kafka из мира обмена сообщениями, какого-то там ивентсорсинга (хотя это тоже не чистая Event Sourcing система, можно накрутить и добавить, есть там определенные решения, которые это все реализуют), люди видят бенефит. И опять же, при добавлении каких-то клевых штук, например, хранилища, которое позволяет хранить всю историю сообщений в системе, которое позволяет в любой момент в любой новой системе перечитать эти данные по новой и получить репорт.

Например, у нас есть система, которая захватывает все заказы, и нам в конце года нужно подвести итоги запустить большую джобу (от англ. job), которая выдаст репорт. Эта джоба запускается один раз в год, но данные нужно откуда-то брать. Чтобы не переливать из одного места в другое, она вычитывает данные из Kafka один раз и получает отчёт.

АНАТОЛИЙ СОЛДАТОВ: Иногда бывают вопросы: у меня в компании RabbitMQ, мы слушали, что Kafka крутая и модная, давайте мы ее поставим на замену, потому что она лучше. Хотя RabbitMQ не хуже в каких-то своих кейсах, и вообще отличная штука для очередей.

ВИКТОР ГАМОВ: Самое лучшее программное обеспечение это то, которым вы умеете пользоваться. Если вы умеете хорошо варить RabbitMQ, и он у вас не падает, по ночам вам не звонят админы, а если вы админ, то не звонят пользователи-разработчики, то вообще нет проблем.

ИВАН СИДОРОВ: Я вот все ждал, когда уровень абстракции поднимется, потому что, вот как я написал про свою должность, от разработчика к архитектору и обратно

В: Ты как Хоббит туда и обратно.

ИВАН СИДОРОВ: Да, и каждый раз с приключениями. Вообще, разница между RabbitMQ и Kafka это то, что одна машина красная, другая синяя. Если в общем говорить. В архитектурном плане. У меня есть теория, откуда начались эти холивары, которые достаточно жестко цепляются за незначительные нюансы. Kafka и RabbitMQ стартовали примерно в одно время. RabbitMQ был написал на Erlang. Тогда был фетиш, все верили, что Erlang это потрясающая система, которая спасёт мир технологий и ускорит всё, а Java это кровавый интерпрайз, мол, чтобы мне запустить на моей VPS-ке с 16-ю мегабайтами памяти Java, придется купить еще 10 таких VPS.

Дальше закон Мура работает, вычислительная мощность позволяет не думать о том, что внутри (если не HFT, а там вычислительные мощности китайских полупроводниковых заводов справляются). Сейчас это абсолютно одинаковые технологии, которые разошлись уже с точки зрения бизнеса.

АНАТОЛИЙ СОЛДАТОВ: Ты вот подводишь, что Kafka и RabbitMQ это одно и то же, но нет. Напомню, на всякий случай.

ИВАН СИДОРОВ: Я говорил с точки зрения архитектуры.

МАРСЕЛЬ ИБРАЕВ: Итак, мы обсудили, что разница у Kafka и RabbitMQ есть. И всегда нужно идти от продукта, от задачи и от умения пользоваться инструментом. Не стоит слепо верить статьям, где написано, что Kafka это круто, и у вас сайтик на WordPress взлетит после её установки.

АНАТОЛИЙ СОЛДАТОВ: Я добавлю ещё про RabbitMQ. Бывает ещё, что ты просто упираешься в технологию. У нас были случаи, когда RabbitMQ приходилось отключать реплику, чтобы он успевал. И это уже звоночек, что надо что-то другое. Плюс всякие истории с нестабильными сетями. Там тоже RabbitMQ разваливается хорошо, и стоит сразу смотреть другие технологии. Даже если вы сейчас умеете с ним работать, то Multi-DC с нестабильными сетями и высокие нагрузки не очень подходят для RabbitMQ.

Как правильно эксплуатировать Kafka

МАРСЕЛЬ ИБРАЕВ: Предположим, мы решили, что проект подходит для Kafka, мы его поставили. Но поставить и настроить это полдела. Дальше его надо эксплуатировать. Как правильно эксплуатировать Kafka? На что стоит обратить внимание? Какие метрики собирать?

АЛЕКСАНДР МИРОНОВ: Надо начать с вопроса, А на чём вы собираетесь запускать Kafka: на виртуальных машинах, на Kubernetes, на bare metal? И от этого способы развёртки и инсталляции будут кардинально меняться. Multi-DC и прочие сетапы добавляют проблем.
Классическая инсталляция Kafka это некий кластер с минимум тремя нодами, чтобы данные копировались как минимум дважды. То есть у вас будет три брокера. Как вы будете их запускать, это зависит от вас: можете с помощью Puppet, можете попробовать сделать это в Kubernetes, что будет намного сложнее. Kafka это распределённая система, у Kafka есть один небольшой недостаток, который постепенно устраняется, это зависимость от ZooKeeper.

ZooKeeper это ещё одна распределённая система, с ещё одним сетом собственных особенностей (мониторинговых, инсталляционных и всего остального), который вам придётся тоже развернуть и использовать.

АНАТОЛИЙ СОЛДАТОВ: Почему недостаток?

АЛЕКСАНДР МИРОНОВ: Ну не знаю, если ты садомазохист, то тебе может быть прикольно запускать две системы, чтобы какого-то продуктового кейса добиться, но вообще

АНАТОЛИЙ СОЛДАТОВ: Ну слушай, ты этот ZooKeeper один раз описываешь в Puppet, и потом тебе уже не важно будет, Kafka одна или с ZooKeeper.

АЛЕКСАНДР МИРОНОВ: Мы же обсуждаем это с точки зрения DevOps? Для любого DevOps-инженера это ещё одна система, которую нужно знать, алертить, мониторить, которая у вас будет падать.

АНАТОЛИЙ СОЛДАТОВ: С другой стороны, ZooKeeper может не только для Kafka использоваться, он вполне может жить в компании в каких-то других системах.

АЛЕКСАНДР МИРОНОВ: А этого, кстати, лучше не делать. Лучше не использовать ZooKeeper для нескольких систем.

АНАТОЛИЙ СОЛДАТОВ: С этим я согласен, но я про другое. Я про то, что ты где-то в Puppet имеешь описанные модули под ZooKeeper, и сетапишь ты его для Kafka или для ClickHouse не важно. Понятно, конфигурация изменится, нагрузка будет другая, но ты уже умеешь с ним работать и у тебя уже мониторинг настроен (немного что-то другое появится, конечно). Я не разделяю нелюбовь к ZooKeeper.

АЛЕКСАНДР МИРОНОВ: Никакой нелюбви нет, просто нужно отдавать себе отчёт, вот и всё

АНАТОЛИЙ СОЛДАТОВ: Да ладно!

АЛЕКСАНДР МИРОНОВ: К слову сказать, в моей команде в Booking.com за всё время эксплуатации Kafka не было каких-то больших проблем с ZooKeeper. Но, чтобы их не было, нам пришлось его подробно изучить.

АНАТОЛИЙ СОЛДАТОВ: ZooKeeper очень старая система. У нас тоже с ним никогда не было проблем.

АЛЕКСАНДР МИРОНОВ: Но у нас в компании всё же были проблемы с ZooKeeper. Не в моей команде, но в других командах. И когда они случаются, они обычно очень серьёзные.

В чём ещё особенность Kafka для людей, которые приходят из managed-сервисов? Например, когда вы работаете с SQS в Amazon, вы чётко знаете лимиты вашей очереди. Вы не можете записать, условно, больше тысячи сообщений в секунду; не можете записать больше чем 1 Мб сообщений в секунду. И вы можете выстраивать приложение, отталкиваясь от этих очень чётких лимитов. С Kafka такого нет. Это зависит и от того, как зависит ваше приложение и продюсеры, и консюмеры; от того, как настроен ваш кластер. Если придёте к админу и скажете: Заведи мне топик и скажи, сколько я смогу записать сообщений в одну партицию. Скорее всего, он не сможет дать вам простой ответ. Это ещё одна особенность Kafka, с которой мы сталкивались очень часто. Потому что приходят люди с очень разным бэкграундом, особенно из продуктовых команд, и последнее, что им нужно это думать про Кб и Мб в секунду, которые они могут пропустить через одну партицию с какой-то latency. Придётся прописывать какие-то гайдлайны для разработчиков. Нужно будет запускать бенчмарки. Нужно будет понимать, вот на вашем сетапе, на вашем железе, сколько данных за какой интервал времени вы можете пропустить.

Что ещё нужно знать про Kafka? Клиенты на каком языке вы пишете, и чем вы пользуетесь. Kafka изначально написана на Scala, теперь уже больше на Java. Соответственно, последний vanilla-клиент, которые поддерживает это всё, это джавовый клиент. Если вы используете C#, то официальный клиент Confluent, например, сделан поверх C-библиотеки librdkafka, которая тоже поддерживается человеком, который работает в Confluent, но всё равно у вас будет задержка с фичами, они будут приходить позже.

Клиенты C Sharp, Java, Go, Python, Ruby мы в Booking.com ещё Perl любили использовать, как вы знаете, нам пришлось свой клиент поверх librdkafka писать. Это достаточно большая боль, которая может возникнуть в тех компаниях, которые используют разные языковые экосистемы.

АНАТОЛИЙ СОЛДАТОВ: Да, но здесь можно либо пользоваться вот этими клиентами или писать свои, либо второе какие-то прокси комплементить или использовать, там уже нет привязки к языку.

АЛЕКСАНДР МИРОНОВ: Совершенно верно! И мы опять приходим к тому, что это ещё один возможный компонент, который придется внедрить, чтобы эту систему развязать. Это всё подчеркивает то, Kafka, она как брокер сообщений, как бы мы его ни называли, достаточно бесполезна сама по себе. Самый большой бенефит Kafka по сравнению с RabbitMQ или Pulsar это её экосистема. Это количество коннекторов, приложений, проприетарных решений, которые просто из коробки будут работать с Kafka. Я знаю даже кейсы из других компаний, где интегрировали данные из нескольких компаний с помощью Kafka. Просто потому, что у них уже она была, и им было намного проще таким образом запродюссить друг другу сообщения в кластере, прокинуть сетевые доступы условные, чем городить свои собственные проприетарные http-протоколы или ещё что-то. Вот именно эта инфраструктура и её распространенность вот это самая главная мощь Kafka, на мой взгляд.

АНАТОЛИЙ СОЛДАТОВ: Мне тоже так кажется. Более того, есть ребята из разных компаний, в том числе российских, которые из Kafka используют только коннекторы. Всё вот это вот связали, поставили Kafka, и за счет этой экосистемы не пишут вообще никакого кода. На конфигах у них всё взлетает, всё работает. Здесь иногда цена внедрения очень маленькая. Она вся упирается в инфраструктуру, по сути.

Чем глубже Kafka прорастает в компанию, есть у нее такая фишка, кластера начинают плодиться с большой силой после первого, инфраструктура тоже разрастается и порой становится очень сложной. Всё что мы перечислили: коннекторы, Zookeeper, репликаторы, мониторинги, штуки, которые позволяют легче администрировать кластера Kafka (потому что администрирование это довольно большой объём toil work). И очень быстро. Стандартная история когда у нас есть Kafka, а вокруг нее еще 10 инфраструктурных сервисов или технологий вращается.

АЛЕКСАНДР МИРОНОВ: На эту тему еще можете посмотреть как раз в Avito мы в январе этого года делали доклады, я и Толя, и мы подробно обсуждали экосистемы. То, как мы делали это в Booking.com и Толя рассказывал, как ребята делают коннекторы в Avito, и можно посмотреть, какое количество дополнительных компонентов нужно для того, чтобы эффективно ранить эту систему, чтобы ваши девелоперы занимались продуктовой разработкой, а не разработкой для Kafka.

Я еще хотел по-быстрому ответить на вопрос из чата: Лучше Kafka Connect или NiFi? В NiFi нет репликации. Если у вас падает нода в NiFi перед тем, как она прокинула данные ещё куда-то, если вы не поднимите её обратно (не знаю, помолитесь или пойдёте диск собирать руками), то не сможете эти данные восстановить.

ИВАН СИДОРОВ: Хочу всё же про мониторинг резюмировать. С коллегами я полностью согласен. Единственная метрика, которую нужно мониторить на инфраструктуре это теряем мы деньги или не теряем, а дальше уже спускаться до самого нижнего компонента. Если рассматривать Kafka как вещь в себе, то её нужно мониторить как обычное Java-приложение. Побольше памяти поставил, она не вылетает всё хорошо.

АЛЕКСАНДР МИРОНОВ: Это, кстати, плохое решение для Kafka, ну да ладно.

ИВАН СИДОРОВ: Мы же в отрыве от инфраструктуры, нам же не сказали данные. А дальше нужно смотреть, выполняет ли выбранная часть инфраструктуры, реализованная на Kafka, свою бизнес-задачу. Самый простой кейс, когда мы используем Kafka как очередь и кидаем с одной стороны сообщение в другое, то смотрим: они отправляются и принимаются? с какой скоростью принимаются? с какой отправляются? почему так мало или много принимается? есть ли аномалии? Аномалии в том бизнес-процессе, который технологически реализует Kafka.

АНАТОЛИЙ СОЛДАТОВ: Ты говоришь про ISO-мониторинг, насколько я услышал, это штука крутая. Кажется, из неё нужно исходить, и всякие там алёрты на пейджеры присылать, когда начинаются всякие такие проблемы, потому что у Kafka отказ одной, пяти или десяти машин это не проблема, можно спать дальше, если всё остальное выполняется.

Но все равно нужно еще иметь инфраструктурный мониторинг, который внутри самой Kafka смотрит, что происходит, потому что когда там проблемы начинаются, в Kafka очень много метрик, это не только то что Java дает, для Kafka их там миллион.

Мы сделали себе три окна. Первое на уровне всех кластеров смотрим, что вообще происходит там, сколько брокеров живых/неживых в процентном соотношении, есть ли офлайн-реплики, например, сколько у нас не выполняется isr.

Дальше спускаемся на уровень конкретного кластера, смотрим, сколько у него входящих сообщений сейчас, сколько исходящих. Сколько было неделю назад в этот же день, в это же время. Количество топиков партиций можно посмотреть, ну, просто серверные метрики, очевидно, в первую очередь надо смотреть сеть, наверное.

АЛЕКСАНДР МИРОНОВ: Сеть, файловые дескрипторы, disk io.

АНАТОЛИЙ СОЛДАТОВ: Файловые дескрипторы можно поставить 1 млн, и не смотреть их. Но у нас они тоже выведены, на самом деле.

АЛЕКСАНДР МИРОНОВ: Ну да, к вам просто придёт 1 млн коннекшенов, вы даже не узнаете об этом.

АНАТОЛИЙ СОЛДАТОВ: Я думаю, мы узнаем об этом быстро. Все пользователи Avito узнают об этом.

И самый низкий уровень это уже когда ты на уровне брокера смотришь специфичные какие-то метрики, сколько там в Q висит это сообщение, как часто меняются shiring expand. Там много всего, что может позволить помочь понять, где проблема (с ДЦ что-то произошло, не вывозят рейды, и их надо увеличить). Такой мониторинг тоже должен быть. Я не вижу кейса, когда вы без него можете понять, что происходит.

ИВАН СИДОРОВ: Есть проблема овермониторинга. Как ты классно рассказал, что вы спускаетесь фактически с SLA-мониторинга на процессы конкретных систем, на конкретный кластер, на конкретную джобу. Если мы спускаемся на конкретную джобу и видим экран из тысячи метрик, и 30 из них горят красными, сделать вывод, что происходит, невозможно. Исчерпывающим мониторингом можно считать тот, который срабатывает в большинстве случаев. А с верхнего уровня у нас стоит все сломалось. И все сломалось запускает ресеч до расширения списков алертов.

АЛЕКСАНДР МИРОНОВ: Мне кажется, что мы просто приходим к топику white box VS black box мониторинг, насколько я знаю у Слёрма есть всякие курсы на эту тему.

АНАТОЛИЙ СОЛДАТОВ: Самый простой путь это взять какой-то проприетарный мониторинг, который работает уже долго (например, от Confluent). Посмотреть, как у них работает, и понять что вам нужно. Там всё довольно-таки красиво, в картинках. Не надо это покупать, но заимплементить через Prometheus, Consul, Grafana, Graphite точно можно.

АЛЕКСАНДР МИРОНОВ: У Kafka в главной документации есть описание критических метрик, за которыми надо следить.

Особенности разработки приложений для работы с Kafka

МАРСЕЛЬ ИБРАЕВ: Есть ли особенности разработки приложения для работы с Kafka? Какие-то подходы?

АЛЕКСАНДР МИРОНОВ: Хочется как-то сформулировать, чтобы опять это не превращать в часовую дискуссию, потому что очень широкие вопросы. Мы уже затронули клиенты. В зависимости от того, какой клиент вы используете, он будет вести себя по-разному. Давайте обсудим Java vanilla клиент, потому что он наверняка самый популярный. В Java vanilla клиенте у вас есть две части этой библиотеки: producers и consumers. Обе эти части тюнятся совершенно по-разному. У каждой свой набор конфигураций, который вы можете настроить в зависимости от того, под какие продуктовые цели используете Kafka. Продюсер вы можете затюнить либо под low latency, чтобы он как можно быстрее отправлял сообщения с высокой гарантией доставки, либо вы можете сказать: мне нужно медленно, но много, или мало, но качественно, либо что-то среднее (in the middle).

По умолчанию, кстати, это касается и настроек брокера тоже, Kafka настроена как система in the middle: она не дает вам супергарантии доставки и не заточена под супер low latency, она где-то посередине, чтобы удовлетворять 80% юзкейсов. Это важно упомянуть, потому что если вы не видите проблем с продюсерами, с консюмерами или брокерами, скорее всего, вам не нужно трогать эти настройки.

Приведу пример. В Booking.com в первые несколько лет мы, может быть, 3 или 4 из сотни top level конфигов меняли. То есть вот этот дефолт действительно очень same в Kafka. Если возвращаться к producer side. Джавовый producer это мультитредное приложение, которое внутри открывает n тредов, коннектится к брокерам, начинает слать сообщения. Лучшая практика его применения это переиспользование одного и того же инстанса в Java. Не нужно открывать тред-сейф, соответственно, вы можете успешно пользоваться одним и тем же инстансом.

С консюмером совершенно другая история, он не тред-сейф. Мало того, что, когда вы делаете какие-то запросы типа poll, если в данный момент в буфере у этого консюмера нет сообщений, которые он вам может отдать напрямую, он заблокирует ваш main trap и начнет делать запросы Kafka. Это уже совершенно два разных поведения вроде как одной и той же библиотеки.
Но я уверен, что в 80-90% вам не надо будет ничего тюнить, вам просто нужно понять, с какой гарантией вы хотите доставить ваше сообщение. Вам, например, без разницы, дойдет оно или нет (поскольку это метрика или log line), или вам важно не потерять его (сообщение о заказе на сайте). Для этого есть одна top level настройка, которая будет фактически контролировать, на какое количество реплик будет записано сообщение, и ответ в ваше приложение вернётся только тогда, когда n реплик подтвердили запись. Соответственно, вы можете сказать, что я просто послал по сети сообщение, и мне даже TCP-акт не нужен, (это акт 0, по-моему).
Про всю эту тему есть отличная статья How to Lose Messages on a Kafka Cluster. И там достаточно большое исследование проведено. Там в Docker чувак поднимал разные конфигурации кластеров, писал в них сообщения, валил эти ноды и смотрел, где какие сообщения теряются. Пойдите почитайте и вы поймёте, что нужно и не нужно делать.

АНАТОЛИЙ СОЛДАТОВ: Я с Сашей полностью согласен. Единственное, добавил бы, что вы, скорее всего, в какой-то момент придёте к коробочкам. У вас будут как раз эти конфиги и вы можете их всех загнать в какой-нибудь фреймворк и использовать всё время одни и те же, чтобы не давать клиентам сильно много ручек, чтобы они не крутили и не портили сами себе жизнь случайно.

АЛЕКСАНДР МИРОНОВ: Согласен. У нас в Booking.com примерно так же и было сделано: ты мог затюнить все конфиги, которые хотел, ничего не ограничивали, но мы давали набор пресетов или коробочек, как ты сказал. У них было human-readable название типа low latency, No AX, high-throughput и мы затюнивали 3-5 параметров в бэкграунд за тебя.

АНАТОЛИЙ СОЛДАТОВ: Да, потому что кажется, что это проблема со всеми системами, где много клиентских настроек, что если у вас сотни сервисов или клиентов, скорее всего, кто-то из них сделает неправильно.

АЛЕКСАНДР МИРОНОВ: Наверное, тут еще про консюмеры нужно упомянуть. Самая главная теория, которую нужно знать про консюмера когда вы отправляете offset, коммитите offset. Гарантия обработки сообщения это: at least once, at most once или exactly once. Это важно для того, чтобы не потерять ваши данные, не потерять результаты обработки ваших данных и, по сути, по большому счету это все сводится к тому, что вот вы считали сообщение, вы получили сообщение из Kafka, после этого вы проводите работу над этим сообщением (неважно, что вы делаете), и перед вами стоит ключевой вопрос: в какой момент вы будете коммитить этот offset обратно в Kafka. Вы будете коммитить его как запроцешенный до того, как вы сделаете процессинг или после? И в зависимости от этого у вас будут разные гарантии доставки и обработки этих сообщений.

АНАТОЛИЙ СОЛДАТОВ: Я бы еще добавил про консюмера то, что он скейлится, их параллелизм зависит напрямую от количества партиций в топике, тоже такая базовая штука. Саша сказал про механику офсетов, есть еще вторая базовая механика: не нужно консюмеров больше делать, чем количество партиций, потому что в Kafka так сделано там больше, чем один консюмер на одну партицию не намапится. Если у вас их будет, условно, 10, а партиций всего три, то никакой пользы вы из этого не извлечете, и вам надо будет сначала партиции увеличить, а потом уже вы получите параллелизм.

АЛЕКСАНДР МИРОНОВ: Совершенно верно, да.

Бэкапы

МАРСЕЛЬ ИБРАЕВ: Последний вопрос из заготовок, который мы успеем рассмотреть: бэкап. Kafka штука распредёленная, и вроде как можно забить и сказать: она сама себя поддерживает, и если что-то упадет, то ну и ладно, она будёт работать. Махнуть рукой и предположить, что никаких проблем не будет. Но при этом что-то такое админское в душе просит все-таки какие-то сохранения состояний, бэкапы на всякий случай. И вопрос такой: нужны ли вообще бэкапы на Kafka, если нужны, то почему или почему не нужны?

АЛЕКСАНДР МИРОНОВ: Я придерживаюсь мнения, что скорее не нужны. Во-первых, в Kafka есть встроенный механизм репликации, который полностью конфигурируется вами, и вы можете контролировать количество реплик от нуля до бесконечности. Помимо этого механизма встроенной серверной репликации у вас есть те самые гарантии доставки, которые мы уже обсудили (тот самый параметр Ex). Вы можете контролировать, на сколько физических или виртуальных машин, контейнеров, подов у вас будет записано сообщение перед тем, как producer получит Всё OK от сервера. Исходя из тех бизнес-кейсов, с которыми работал я, этой гарантии было более чем достаточно.

Тут ещё такой момент, что Kafka имеет замечательный встроенный механизм ретеншена данных. Она может из коробки по времени или по размеру вашего топика удалять старые сегменты. Сегменты данных, которые были записаны, допустим, неделю назад, если вы ее так сконфигурировали. Это даёт вам гарантию, что сейчас внутри вашего кластера будут данные за прошедшую неделю, а более ранние Kafka автоматически из кластера выкинет. Чаще всего Kafka используется именно с таким паттерном, а не как бесконечная база данных.

Один из популярных паттернов складывать результаты аналитической обработки в тот же самый Amazon S3, чтобы потом их можно было поднять другим каким-то компонентом.

Ещё один интересный момент это поддержка infinite retention, то есть бесконечного ретеншена. Это значит, что вы сможете сконфигурировать Kafka таким образом, что вот те самые старые сегменты, которые выкидываются и просто удаляются с диска, вы можете сказать Kafka, что она должна их сложить в какой-то холодный object store (по большому счёту, это Tiered Storage). Интерпрайзный уже есть, но я им не пользовался, так что не буду говорить (open source должен появиться). Так что холодные сегменты, данные, которые вы уже не будете активно использовать, вы можете сложить в S3 и потом доставать раз в год, когда нужно перечитать весь лог. Так вы будете сохранять место на горячем диске и при этом не понадобиться задумываться о бэкапах и прочих дополнительных настройках.

АНАТОЛИЙ СОЛДАТОВ: Я полностью за, и у меня есть несколько дополнений. Начну с Tiered Storage. Это крутая штука, которая изменит в целом конфигурацию машин по Kafka. Сейчас у нас есть какие-то заряженные железками, дисками машины, а можно будет просто какие-то маленькие (почти in memory) Kafka ставить, а все остальное складывать в Tiered Storage. Это как один из вариантов использования.

По поводу бэкапов у нас такой же подход. Kafka это по умолчанию распределённая отказоустойчивая система, которая не должна терять данные, и она для этого практически всё делает. Угробить можно любую систему, конечно. Один из ключевых конфигов, которые нужно не забыть использовать это rack awareness, на мой взгляд. Даже если вы живете в одном дата-центре, попробуйте, если у вас есть контроль над серверами, хотя бы в разные стойки разнести брокеры, всем брокерам прописать, где они стоят (rack id). И Kafka будет следить, чтобы у вас все лидеры равномерно по этим рэкам распределились. Тогда вероятность проблем минимальная. При этом стоимость бэкапов Kafka высокая. Так что Kafka бэкапить-то можно, но будет это стоить дорого, а реального профита вы наверняка не получите.

И второй момент: у Kafka есть Zookeeper, как мы уже обсуждали, и это та система, которую, бэкапить можно. Она довольно недорогая. Мы бэкапим ее. Просто раз в неделю снимаем бэкапы, валидируем их довольно примитивно. Смотрим, что у брокеров id стоят, в Docker разворачиваем и считаем, что всё OK.

АЛЕКСАНДР МИРОНОВ: Мы тоже бэкапим Zookeeper, это важно.

АНАТОЛИЙ СОЛДАТОВ: Без Zookeeper Kafka теряет информацию о том, какие данные где лежат, как их получить. Даже если вся Kafka работает хорошо, а кластер Zookeeper по каким-то причинам отлетел, вам будет очень сложно восстановить данные. Есть всякие Kafka-дампы, которыми можно вытащить из конкретных топиков, которые вы знаете, где лежат, конкретные данные. Но это будет долго и скорее всего не восстановите весь кластер. Поэтому лучше Zookeeper-мозги где-то хранить.

АЛЕКСАНДР МИРОНОВ: При том что объём этих мозгов измеряется в Мб.

Там еще спрашивают в чате: учитывается ли rack id в min.insync.replicas? Я отвечу нет. Просто смысл rack id, в том, что он используется Kafka в момент создания топика. Когда вы говорите, сколько у вас есть партиций в топике, и для каждой у вас будет n реплик. Вот этот rack id, Kafka вам гарантирует, что эти реплики партиций будут раскиданы по разным рэкам. Что она не положит их все в один рэк. После этого она и никакой rack id в своём рантайме не использует.

АНАТОЛИЙ СОЛДАТОВ: Не Kafka использует, а всякие админ-команды. Это будет влиять только на ребалансировки, когда вы будете партиции двигать по брокерам. Тогда да. Он опять заиграет, и Kafka будет смотреть на них так, чтобы лидеров или партиций перетащить так, чтобы они были во всех стойках были распределены правильно.

АЛЕКСАНДР МИРОНОВ: И если, например, вы зададите, что у вас два рэка, но при этом min.insync.replicas у вас там три, то по умолчанию команда создания топиков скажет вам, что не может создать топик, потому что не может раскидать партиции равномерно.

ИВАН СИДОРОВ: По опыту наших клиентов на поддержке могу сказать, что сейчас данные, хранимые в Kafka, не настолько дороги, чтобы делать бэкапы с них. Пока что в типовых случаях время жизни данных в Kafk довольно короткое. Тут правильная мысль про то, что реплики должны быть в разных дата-центрах. Но реплика это кластер, и даже учитывая, что мастер-ноды там нет, они сами себе выбирают, с кластером что-то может пойти не так. Но сохранять данные очень дорого по соотношению стоимость данных/стоимость хранения данных, и самый эффективный вариант, который мы видели это mirroring кластера. Чем отличается от реплики: поднимается точно такой же кластер и настраивается туда пересылка данных. Он просто работает как резерв и как бэкап.

АНАТОЛИЙ СОЛДАТОВ: Тут сразу два момента интересных если несколько ДЦ, это не синоним того, что у нас несколько кластеров. И всегда, когда про Kafka читаешь, начинается с того, что О! Stretch cluster, ужас какой, не используйте. Это всё неправда. На самом деле, нужно исходить из latency, которую ваша сеть дает, проверить хотя бы это, и скорее всего там stretch cluster заведётся, будет всё работать хорошо. Есть варианты с асинхронными кластерами, которые на основе репликаторов, но здесь тоже не стоит забывать, что это повышает сложность всей системы и то, что появляется еще одна точка отказа, и это не бесплатно. И начинать все-таки стоит с простого stretch cluster, который один, растянутый на несколько дата-центров. И часто он работает.

ИВАН СИДОРОВ: Добавлю ещё не про кластеры, а про mirroring кластера. Читал в вопросах, как разделять контуры безопасности, практический опыт если есть интранет и экстранет, и в интранет пускать нельзя, а потоки данных идут из внешки, можно миррорить часть топиков из внешней Kafka во внутреннюю Kafka. Открыть только одну дырочку.

Для тех, кто хочет узнать больше о настройке и оптимизации Apache Kafka Слёрм готовит новый видеокурс. Спикеры Анатолий Солдатов из Авито и Александр Миронов из Stripe. Доступ к первым урокам будет бесплатным. Оставить заявку на курс можно уже сейчас.

Мы открыли доступ к базовым темам курса по Apache Kafka, начать учиться можно уже сейчас.

В программе две теоретические темы Введение и Базовые основы технологии и практическая тема Установка Kafka. В ней поработаем с технологией руками:

1. Развернём Kafka в самом простом её варианте с одним брокером и одной нодой ZooKeeper.
2. Запишем и прочитаем сообщения, посмотрим в конфиги и увидим, как данные хранятся на диске.

Спикеры курса

• Анатолий Солдатов, Lead Engineer в Авито;
• Александр Миронов, Infrastructure Engineer в Stripe, ex-Booking.

Доступ к курсу придёт после регистрации: https://slurm.io/kafka

Релиз всех тем видеокурса будет 7 апреля. До релиза действует цена предзаказа 40 000 руб. вместо 50 000 руб.

В Kafka топик может содержать множество партиций, между которыми распределяются записи. Партиции это единицы параллелизма. В целом, чем больше партиций, тем выше пропускная способность. Однако есть некоторые факторы, которые стоит учитывать, когда в кластере Kafka много партиций. Рад сообщить, что в Apache Kafka, начиная с версии 1.1.0, было существенно увеличено количество партиций, которые может поддерживать один кластер Kafka с точки зрения деплоймента и доступности.

Чтобы понять это улучшение, будет полезно повторить базовую информацию о лидерах партиций и контроллере. Во-первых, каждая партиция может иметь несколько реплик для большей доступности и устойчивости. Одна из реплик назначается лидером, и все запросы клиента обслуживаются этой ведущей репликой. Во-вторых, один из брокеров в кластере выступает в качестве контроллера, который управляет всем кластером. В случае отказа брокера контроллер отвечает за выбор новых лидеров партиций на отказавшем брокере.

Брокер Kafka по умолчанию выполняет контролируемое отключение, чтобы минимизировать сбои в обслуживании клиентов. Контролируемое отключение проходит следующие этапы. (1) Сигнал SIG_TERM отправляется брокеру для завершения работы. (2) Брокер отправляет запрос контроллеру, уведомляя, что он готов к отключению. (3) Контроллер затем меняет лидеров партиций на этом брокере на других брокеров и сохраняет эту информацию в ZooKeeper. (4) Контроллер отправляет информацию о новых лидерах другим брокерам в кластере. (5) Контроллер отправляет выключающемуся брокеру положительный ответ на запрос, и брокер, наконец, завершает свой процесс. Таким образом это никак не влияет на клиентов, потому что их трафик уже перенаправлен на других брокеров. Этот процесс изображен на Рисунке 1. Заметьте, что шаги (4) и (5) могут происходить параллельно.

Рис. 1. (1) Инициация отключения на брокере 1; (2) брокер 1 отправляет запрос о контролируемом отключении контроллеру на брокере 0; (3) контроллер записывает новых лидеров в ZooKeeper; (4) контроллер отправляет информацию о новых лидерах брокеру 2; (5) контроллер отправляет положительный ответ брокеру 1.

До версии Kafka 1.1.0 во время контролируемого отключения контроллер перемещает лидеров по одной партиции за раз. Для каждой партиции контроллер выбирает нового лидера, параллельно записывает его в ZooKeeper и отправляет нового лидера другим брокерам через удаленный запрос. В этом процессе есть несколько недостатков. Во-первых, у синхронной записи в ZooKeeper более высокое время задержки, что замедляет процесс контролируемого отключения. Во-вторых, отправка нового лидера по партиции за раз добавляет много маленьких удаленных запросов к каждому брокеру, что может вызвать задержку обработки новых лидеров.

Начиная с Kafka 1.1.0 и далее контролируемое отключение происходит значительно быстрее. Во-первых, при записи в ZooKeeper используется асинхронный API. Вместо того чтобы записывать лидера для одной партиции, ждать, пока этот процесс завершится, а потом записывать следующего, контроллер асинхронно размещает лидеров для многих партиций в ZooKeeper, а потом ждет завершения процесса. Это позволяет обрабатывать запросы на пайплайне между брокером Kafka и сервером ZooKeeper и снижает общее время задержки. Во-вторых, информация о новых лидерах отправляется батчами. Вместо одного удаленного запроса для каждой партиции контроллер отправляет один удаленный запрос с лидером от всех затронутых партиций.

Время обработки отказа контроллером также было значительно улучшено. Если случается отказ контроллера, кластер Kafka автоматически выбирает контроллером другого брокера. Прежде чем иметь возможность выбрать лидеров партиций свеженазначенный контроллер должен загрузить состояние всех партиций в кластере из ZooKeeper. Если произошел серьезный сбой контроллера, окно недоступности партиции может длиться столько, сколько продолжается срок действия сессии ZooKeeper, плюс время перезагрузки состояния контроллера. Поэтому сокращение времени перезагрузки состояния улучшает доступность в таком редком случаем. До версии 1.1.0 в Kafka для перезагрузки используется синхронный API ZooKeeper. Начиная с версии 1.1.0 это также заменено на асинхронный API для сокращения времени задержки.

Мы провели тесты для оценки улучшений по времени контролируемого отключения и перезагрузки контроллера. Для обоих тестов мы загрузили набор из пяти узлов ZooKeeper на выделенных серверах.

Для первого теста мы подготовили кластер Kafka с пятью брокерами на отдельных серверах. В этом кластере мы создали 25 000 топиков, в каждом топике по одной партиции и две реплики, в общем получив 50 000 партиций. Таким образом на каждом брокере было 10 000 партиций. Затем мы замерили время, которое понадобилось для контролируемого отключения. Результаты представлены в таблице ниже.

Большую часть улучшений дает исправление затрат на журналирование, при котором проводятся ненужные записи для каждой партиции в кластере при смене лидера одной партиции. Просто исправив затраты на журналирование, мы снизили время контролируемого отключения с 6,5 минут до 30 секунд. Переход на асинхронный API ZooKeeper снизил это время до 3 секунд. Эти улучшения существенно снизили время, необходимое для перезагрузки кластера Kafka.

Для второго теста мы подготовили другой кластер Kafka, состоящий из пяти брокеров, создали 2 000 топиков, в каждом по 50 партиций и одной реплике. В сумме во всем кластере получилось 100 000 партиций. Затем мы замерили время перезагрузки состояния контроллера и увидели 100% улучшение (время перезагрузки снизилось с 28 секунд в Kafka 1.0.0 до 14 секунда в Kafka 1.1.0).

Учитывая эти изменения, на поддержку какого количества партиций вы можете рассчитывать в Kafka? Точное число зависит от таких факторов как допустимое окно недоступности, время задержки ZooKeeper, тип хранения на брокере и т.д. В качестве общего правила мы рекомендуем иметь на каждом брокере до 4 000 партиций и до 200 000 партиций в кластере. Основная причина для лимита на кластере заключается в том, что нужно подстраховаться на тот редкий случай серьезного сбоя контроллера, о котором мы писали выше. Обратите внимание, что другие соображения, связанные с партициями, также применимы, и вам может потребоваться дополнительная настройка конфигурации с большим количеством партиций.

Более подробную информацию вы найдете в KAFKA-5642 и KAFKA-5027.

От редакции: Команда Слёрма готовит подробный видеокурс по Apache Kafka. Спикеры Анатолий Солдатов из Авито и Александр Миронов из Stripe. Вводные уроки уже доступны в личном кабинете. Релиз полной версии курса апрель 2021. Подробности.

Pulsar или Kafka что лучше? Здесь мы обсудим плюсы и минусы, распространенные мифы и нетехнические критерии, чтобы найти лучший инструмент для ваших задач.

Обычно я рассказываю об Apache Kafka и ее экосистеме. О Pulsar за последние годы меня спрашивали только коммитеры и авторы Pulsar. Они задавали сложные технические вопросы, чтобы показать, что Kafka не идет ни в какое сравнение с Pulsar. На Reddit и подобных платформах разгораются яростные и очень субъективные споры на эту тему. Я поделюсь своей точкой зрения, основанной на многолетнем опыте работы со стриминговыми опенсорс-платформами.

Сравнение технологий нынче в моде: Kafka vs. Middleware, Event Streaming и API Platforms

Цель сравнения технологий помочь людям подобрать подходящее решение и архитектуру для своих потребностей. Универсального решения не существует. И это не дело вкуса. Просто для каждой задачи есть свой инструмент.

При этом такие сравнения почти всегда субъективны. Даже если автор не работает на вендора и представляется независимым консультантом, скорее всего, у него все равно есть любимчики, даже если он сам себе в этом не признается. Однако полезно посмотреть на инструменты с разных точек зрения. Поскольку Apache Pulsar сейчас обсуждают, я решил поделиться своим мнением и сравнить его с Kafka. Я работаю в Confluent, а мы лучшие эксперты по Apache Kafka и ее экосистеме. И все же я постараюсь быть максимально объективным и оперировать голыми фактами.

Опенсорс-фреймворки и коммерческие программы постоянно сравнивают. Я и сам делал такие сравнения в своем блоге и на других платформах, например InfoQ: Сравнение платформ интеграции, Выбор подходящего ESB для ваших потребностей, Kafka vs. ETL / ESB / MQ, Kafka vs. Mainframe и Apache Kafka и API Management / API Gateway. Просто заказчики хотели понять, какой инструмент поможет решить те или иные задачи.

Если говорить о сравнении Pulsar и Kafka, ситуация немного отличается.

Зачем сравнивать Pulsar и Kafka?

Реальные и потенциальные заказчики редко спрашивают о Pulsar. Хотя, если честно, в последние месяцы чуть чаще где-то на каждой 15-й или 20-й встрече. Потому что фичи и варианты использования у них частично совпадают. Хотя, по-моему, все дело в паре статей о том, что Pulsar якобы в чем-то лучше Kafka. В пользу противоположной точки зрения нет никаких фактов и очень мало материала.

Я не видел ни одной организации, которая бы всерьез задумывалась о Pulsar в продакшене, хотя по всему миру множество пользователей, которым нужна технология распределенной обработки сообщений, вроде Kafka или Pulsar. По-моему, те, кто рекомендует Pulsar, чего-то не договаривают.

Например, их главный пользователь Tencent, крупная китайская технологическая компания, которая при этом очень активно использует Kafka везде, кроме одного проекта, где пригодился Pulsar. Tencent обрабатывает с Kafka десять триллионов сообщений в день (только представьте: 10 000 000 000 000). Если посчитать, Tencent использует Kafka в тысячу раз активнее, чем Pulsar (10 трлн против десятков млрд). Tencent с удовольствием рассказывают о своем деплое Kafka: Как Tencent PCG использует Apache Kafka для обработки 10+ трлн сообщений в день.

Сравнение двух конкурирующих опенсорс-платформ

Apache Kafka и Apache Pulsar две замечательные конкурирующие технологии. Логично будет их сравнить.

У Apache Kafka и Apache Pulsar очень схожий набор функций. Советую оценить обе платформы на предмет доступных функций, зрелости, распространения на рынке, опенсорс-инструментов и проектов, обучающего материала, проведения митапов по регионам, видео, статей и так далее. Примеры вариантов использования в вашей отрасли помогут принять правильное решение.

Confluent уже писал сравнение Kafka vs. Pulsar vs. RabbitMQ: производительность, архитектура и функции. Я тоже поучаствовал. Значит, сравнение уже есть

О чем тогда эта статья?

Я хочу рассмотреть несколько мифов из споров на тему Kafka против Pulsar, которые регулярно возникают в блогах и на форумах. Затем я сравню их не только по техническим аспектам, потому что обычно никто не говорит о Pulsar с этой стороны.

Kafka vs Pulsar мифы

Мне попадалось несколько мифов. С некоторыми я согласен, другие можно легко развенчать фактами. Ваше мнение может отличаться от моего, это нормально. Я излагаю исключительно свою точку зрения.

Миф 1. У Pulsar есть характеристики, которых нет у Kafka.

Правда.

Если сравнивать Apache Kafka и Apache Pulsar, можно найти различия в многоуровневой архитектуре, очередях и мультитенантности.

Но!

У Kafka тоже есть уникальные особенности:

• В два раза меньше серверов, которыми приходится управлять.
• Данные сохраняются на диск только один раз.
• Данные кэшируются в памяти только однажды.
• Проверенный протокол репликации.
• Производительность zero-copy.
• Транзакции.
• Встроенная обработка потока.
• Долгосрочное хранилище.
• В разработке: удаление ZooKeeper (KIP-500), чтобы использовать и деплоить Kafka было еще проще, чем Pulsar с его четырехкомпонентной архитектурой (Pulsar, ZooKeeper, BookKeeper и RocksDB), а еще чтобы повысить масштабируемость, устойчивость и т. д.
• В разработке: многоуровневое хранилище (KIP-405), чтобы повысить эластичность и экономичность Kafka.

Спросите себя: можно ли сравнивать опенсорс-платформы или продукты и вендоров с комплексным предложением?

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

Например. Понадобилось несколько лет, чтобы реализовать и испытать в бою Kafka Streams в качестве нативного движка обработки потоков. Как это можно сравнивать с Pulsar Functions? Последнее позволяет добавлять определяемые пользователем функции (UDF) безо всякой связи с обработкой реальных потоков. Или это скорее похоже на Single Message Transformations (SMT), основную функцию Kafka Connect? Не сравнивайте круглое с квадратным и не забывайте учитывать зрелость. Чем критичнее функция, тем больше зрелости ей требуется.

Сообщество Kafka вкладывает невероятные усилия в работу над основным проектом и его экосистемой. Только в Confluent 200 инженеров занимаются исключительно проектом Kafka, дополнительными компонентами, коммерческими продуктами и предложениями SaaS в основных облачных провайдерах.

Миф 2. У Pulsar есть несколько крупных пользователей, например китайский Tencent.

Правда.

Но! Tencent использует Kafka активнее, чем Pulsar. Отдел выставления счетов, где используется Pulsar, это лишь малая часть Tencent, в то время как остальные используют Kafka. У них есть глобальная архитектура на основе Kafka, где больше тысячи брокеров сгруппированы в единый логический кластер.

Будьте осторожны с опенсорс-проектами. Проверяйте, какого успеха добились с ними обычные компании. Если эту технологию использует один технологический гигант, это еще не значит, что вам она тоже подойдет. Сколько компаний из Fortune 2000 могут похвастаться историями успеха с Pulsar?

Ищите примеры не только среди гигантов!

Примеры обычных компаний позволят лучше понять, как применяется тот или иной инструмент в реальном мире. Если это не истории успеха от самих вендоров. На сайте Kafka можно найти много примеров компаний. Более того, на конференциях Kafka Summit в Сан-Франциско, Нью-Йорке и Лондоне каждый год разные предприятия из разных отраслей делятся своими историями и кейсами. Это компании из Fortune 2000, предприятия среднего бизнеса и стартапы.

Приведу один пример про Kafka. Для репликации данных в реальном времени между отдельными кластерами Kafka существует много разных инструментов, включая MirrorMaker 1 (часть проекта Apache Kafka), MirrorMaker 2 (часть проекта Apache Kafka), Confluent Replicator (от Confluent, доступен только в рамках Confluent Platform и Confluent Cloud), uReplicator (опенсорс от Uber), Mirus (опенсорс от Salesforce), Brooklin (опенсорс от LinkedIn).

На практике разумно использовать только два варианта, если, конечно, вы не хотите обслуживать и улучшать код самостоятельно. Это MirrorMaker 2 (еще не вполне зрелая новинка, но отличный выбор для средне- и долгосрочной перспективы) и Confluent Replicator (проверенный на практике во многих критически важных для бизнеса деплоях, но платный). Остальные варианты тоже нормальные. Но кто обслуживает эти проекты? Кто разбирается с багами и проблемами безопасности? Кому звонить, если в продакшене что-то случилось? Одно дело деплоить критически важные системы в продакшене, другое оценивать и пробовать опенсорс-проект.

Миф 3. Pulsar предлагает очереди и стриминг в одном решении.

Частично.

Очереди используются для обмена сообщениями между двумя точками. Они предоставляют асинхронный протокол взаимодействия, то есть отправитель и получатель сообщения не должны обращаться к очереди одновременно.

Pulsar предлагает только ограниченную поддержку очередей сообщений и стриминга событий. В обеих областях он пока не может составить крепкую конкуренцию по двум причинам:

1. Pulsar предлагает ограниченную поддержку очередей сообщений, потому что ему не хватает таких популярных функций, как XA-транзакции, маршрутизация, фильтрация сообщений и т. д. Это обычные функции в таких системах сообщений, как IBM MQ, RabbitMQ, и ActiveMQ. Адаптеры Pulsar для систем сообщений тоже ограничены в этом смысле. В теории все выглядит нормально, но на практике...

   
   

2. Pulsar предлагает ограниченную поддержку стриминга. Например, на практике в большинстве случаев он не поддерживает семантику строго однократной (exactly-once) доставки и обработки. Вряд ли кто-то станет использовать Pulsar для платежной системы, потому что платежи могут дублироваться и теряться. Ему не хватает функционала для обработки потоков с соединением, агрегированием, окнами, отказоустойчивым управлением состоянием и обработкой на основе времени событий. Топики в Pulsar отличаются от топиков в Kafka в худшую сторону из-за BookKeeper, который был придуман в 2008 году как лог упреждающей записи для HDFS namenode в Hadoop в расчете на краткосрочное хранение.

   
   

Примечание. Адаптер Kafka для Pulsar тоже ограничен. В теории звучит заманчиво, но в реальности не все получается, потому что он поддерживает только небольшую часть функционала Kafka.

Как и Pulsar, Kafka предлагает ограниченную поддержку очереди.

В Kafka можно использовать разные обходные пути, чтобы реализовать нормальное поведение очереди. Если вы хотите использовать отдельные очереди вместо общих топиков Kafka для:

• Безопасности => используйте Kafka ACL (и дополнительные инструменты, вроде контроля доступа на основе ролей (RBAC) от Confluent).
• Семантики (отдельных приложений) => используйте группы консюмеров в Kafka.
• Балансировки нагрузки => используйте партиции Kafka.

Обычно я спрашиваю заказчиков, зачем им нужны очереди. Kafka предлагает готовые решения для разных сценариев, на которые можно просто посмотреть с другой стороны. Редко встречаются случаи, когда действительно требуется высокая пропускная способность с очередями.

Зная обходные пути и ограничения Pulsar и Kafka в плане сообщений, давайте проясним: ни одна из платформ не предлагает решение для обмена сообщениями.

Если вам нужно именно оно, возьмите что-то вроде RabbitMQ или NATS.

Однозначного решения здесь нет. Многие наши заказчики заменяют имеющиеся системы сообщений, вроде IBM MQ, на Kafka (из соображений масштабируемости и затрат). Просто нужно знать имеющиеся варианты и их недостатки, все взвесить и подобрать решение, которое лучше всего подходит именно вам.

Миф 4. Pulsar предоставляет потоковую обработку.

Неправда.

Или, если по-честному, все зависит от того, что вы называете обработкой потоков. Это какая-то простейшая функция или прямо полноценная обработка?

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

Pulsar предлагает минимальную потоковую обработку через Pulsar Functions. Она подходит для простых обратных вызовов, но не сравнится с функционалом Kafka Streams или ksqlDB для создания стриминговых приложений со stateful-информацией, скользящими окнами и другими функциями. Варианты применения можно найти в разных отраслях. Например, на сайте Kafka Streams есть кейсы New York Times, Pinterest, Trivago, Zalando и других.

Для стриминга аналитики Pulsar обычно используется в сочетании с нормальной платформой обработки потоков, вроде Apache Spark или Apache Flink, но вам придется управлять дополнительным элементами распределенной архитектуры и разбираться в их сложном взаимодействии.

Миф 5. Pulsar предоставляет семантику exactly-once, как и Kafka.

Неправда.

В Pulsar есть функция дедупликации, чтобы одно сообщение не сохранялось в брокере Pulsar дважды, но ничто не мешает консюмеру прочитать это сообщение несколько раз. Этого недостаточно для любого применения обработки потоков, где для входа и выхода используется Pulsar.

В отличие от функции транзакций в Kafka, здесь нельзя привязывать отправленные сообщения к состоянию, записанному в обработчике потоков.

Семантика Exactly-Once Semantics (EOS) доступна с версии Kafka 0.11, которая вышла три года назад, и используется во многих продакшен-деплоях. Kafka EOS поддерживает всю экосистему Kafka, включая Kafka Connect, Kafka Streams, ksqlDB и такие клиенты, как Java, C, C++, Go и Python. На конференции Kafka Summit было несколько выступлений, посвященных функционалу Kafka EOS, включая это превосходное и понятное введение со слайдами и видео.

Миф 6. У Pulsar производительность выше, чем у Kafka.

Неправда.

Я не поклонник разных бенчмарков производительности и пропускной способности. Обычно они субъективны и относятся к конкретной задаче независимо от того, кто их проводит вендор, независимый консультант или исследователь.

Например, GIGAOM опубликовали один бенчмарк, где сравнивается задержка и производительность в Kafka и Pulsar. Автор намеренно замедлил Kafka, настроив параметр flush.messages = 1, в результате которого каждый запрос вызывает fsync и Kafka синхронизируется с диском при каждом сообщении. В этом же бенчмарке консюмер Kafka отправляет подтверждение синхронно, а консюмер Pulsar асинхронно. Неудивительно, что Pulsar вышел явным победителем. Правда, автор забыл упомянуть или объяснить существенные отличия в конфигурации и измерениях. Давайте не будем путать теплое с мягким.

На самом деле архитектура Pulsar сильнее нагружает сеть (из-за брокера, который выступает как прокси перед серверами bookie BookKeeper) и требует в два раза больше пропускной способности (потому что BookKeeper записывает данные в лог упреждающей записи и в главный сегмент).

Confluent тоже делал бенчмарки. Только там сравнивалось сопоставимое. Неудивительно, что результаты получились другими. Но нужно ли вообще встревать в эти бои на бенчмарках между вендорами?

Оцените свои требования к производительности. Сделайте proof-of-concept с Kafka и Pulsar, если надо. Скорее всего, в 99% случаев обе платформы покажут приемлемую производительность для вашего сценария. Не доверяйте посторонним субъективным бенчмаркам! Ваш случай все равно уникален, и производительность это только один из аспектов.

Миф 7. Pulsar проще в использовании, чем Kafka.

Неправда.

Без дополнительных инструментов вам будет сложно и с Kafka, и с Pulsar.

У Kafka две распределенных системы: сама Kafka и Apache ZooKeeper.

Но! У Pulsar три распределенных системы, а еще хранилище: Pulsar, ZooKeeper и Apache BookKeeper. Как и Pulsar, BookKeeper использует ZooKeeper. Для некоторых задач хранения используется RocksDB. Это означает, что Pulsar гораздо сложнее понять и настроить по сравнению с Kafka. Кроме того, у Pulsar больше параметров конфигурации, чем у Kafka.

Kafka движется в противоположном направлении скоро ZooKeeper будет удален (см. KIP-500), так что останется всего одна распределенная система, которую нужно деплоить, обслуживать, масштабировать и мониторить:

ZooKeeper мешает масштабированию в Kafka и усложняет эксплуатацию. Но у Pulsar все еще хуже!

Одна из главных проблем моих заказчиков использование ZooKeeper в критически важных деплоях в большом масштабе. С нетерпением жду упрощения архитектуры Kafka, чтобы остались только брокеры. Кроме того, мы получим единую модель безопасности, потому что больше не придется отдельно настраивать безопасность для ZooKeeper. Это огромное преимущество, особенно для крупных организаций и отраслей со строгим регулированием. Ребята из комплаенса и ИБ скажут вам спасибо за упрощенную архитектуру.

Использование платформы связано НЕ только с архитектурой

У Kafka гораздо более подробная документация, огромное сообщество экспертов и большой набор поддерживающих инструментов, которые упрощают работу.

Кроме того, по Kafka есть много вариантов обучения онлайн-курсы, книги, митапы и конференции. К сожалению, у Pulsar с этим все плохо.

Миф 8. Архитектура с тремя уровнями лучше, чем с двумя.

Зависит от ситуации.

Лично я скептически отношусь к тому, что трехуровневую архитектуру Pulsar (брокеры Pulsar, ZooKeeper и BookKeeper) можно считать преимуществом для большинства проектов. Тут есть издержки.

Twitter рассказал, как отказался от BookKeeper + DistributedLog (DistributedLog похож на Pulsar по архитектуре и дизайну) около года назад, соблазнившись такими преимуществами одноуровневой архитектуры Kafka, как экономичность и улучшенная производительность, которых недоставало двухуровневой архитектуре с отдельным хранилищем.

Как и Pulsar, DistributedLog построен на BookKeeper и добавляет стриминговый функционал, схожий с Pulsar (например, уровень обработки существует отдельно от уровня хранилища). Изначально DistributedLog был отдельным проектом, но потом присоединился к BookKeeper, хотя сегодня им, похоже, мало кто занимается (всего пара коммитов за последний год). Среди основных причин перехода Twitter на Kafka называлась значительная экономия и повышение производительности, а также обширное сообщество и популярность Kafka. Вот какие выводы они сделали: Для одного консюмера экономия ресурсов составила 68%, а для нескольких целых 75%.

Преимущества трехуровневой архитектуры проявляются при создании масштабируемой инфраструктуры. При этом дополнительный уровень увеличивает нагрузку на сеть минимум на 33%, а данные, которые хранятся в брокерах Pulsar, приходится дополнительно кэшировать на обоих уровнях, чтобы добиться аналогичной производительности, а еще дважды записывать на диск, потому что формат хранения в Bookkeeper основан не на логе.

В облаке, где работает большинство деплоев Kafka, лучшим внешним хранилищем выступает не такая узкоспециализированная технология, как BookKeeper, а популярное и проверенное хранилище объектов, вроде AWS S3 и GCP GCS.

Tiered Storage в Confluent Platform на основе AWS S3, GCP GCS и им подобных, дает те же преимущества без дополнительного слоя BookKeeper, как у Pulsar, и без дополнительных затрат и задержек, связанных с передачей данных по сети. У Confluent ушло два года на то, чтобы выпустить GA-версию Tiered Storage for Kafka с круглосуточной поддержкой для самых критичных данных. Tiered Storage пока недоступно для опенсорсной версии Apache Kafka, но Confluent вместе с сообществом Kafka (включая крупные технологические компании, вроде Uber) работает над KIP-405, чтобы добавить Tiered Storage в Kafka с разными вариантами хранения.

У обеих архитектур есть плюсы и минусы. Лично мне кажется, что 95% проектов не нуждаются в сложной трехуровневой архитектуре. Она может пригодиться там, где требуется внешнее недорогое хранилище, но вам придется отвечать за круглосуточное соблюдение SLA, масштабируемость и пропускную способность. А еще за безопасность, управление, поддержку и интеграцию в вашу экосистему. Если вам действительно нужна трехуровневая архитектура, не стану вас останавливать.

Связанный миф: Pulsar лучше подходит для отстающих консюмеров благодаря уровню кэширования и хранения.

Неправда.

Основная проблема отстающих консюмеров в том, что они занимают page cache, то есть последние сообщения уже кэшированы. При чтении из предыдущих сегментов они заменяются, что снижает производительность консюмеров, читающих с начала лога.

Архитектура Pulsar проявляет себя даже хуже в этом отношении. Проблема сброса кэша сохраняется, при этом для чтения нужно сделать лишний прыжок по сети и загрузить сеть вместо того, чтобы просто прочитать сообщение с локального носителя.

Миф 9. Kafka масштабируется хуже, чем Pulsar.

Неправда.

Это один из главных аргументов сообщества Pulsar. Как я уже говорил, это всегда зависит от выбранного бенчмарка. Например, я видел тесты с эквивалентными вычислительными ресурсами, где на высокой пропускной способности у Kafka результат был гораздо лучше, чем у Pulsar. Вот один из таких бенчмарков:

В большинстве случаев масштабируемость не проблема. Kafka можно легко разогнать до нескольких гигабайтов в секунду, как в демонстрации Масштабирование Apache Kafka до 10+ ГБ в секунду в Confluent Cloud:

Честно говоря, этот вопрос должен волновать менее 1% пользователей. Если у вас требования, как у Netflix (петабайты в день) или LinkedIn (триллионы сообщений), есть смысл обсуждать самую подходящую архитектуру, железо и конфигурацию. Остальным можно не беспокоиться.

Связанный миф: сейчас в Kafka может храниться всего 500к партиций на кластер.

Правда.

Пока у Kafka не лучшая архитектура для масштабных деплоев с сотнями тысяч топиков и партиций.

Но! Pulsar тоже не резиновый. Просто у него другие лимиты.

Ограничения в Kafka связаны с Zookeeper. Когда Zookeeper удалят через KIP-500, верхняя граница уйдет вместе с ним.

Примечание: успех зависит от подходящего устройства архитектуры!

Если у заказчика возникли проблемы с количеством партиций и масштабируемостью в Kafka, скорее всего, он ошибся в архитектуре и приложениях (Pulsar бы тут точно не помог).

Kafka это платформа стриминга, а не очередной IBM MQ. Если вы попробуете воссоздать любимое решение и архитектуру MQ с Kafka, у вас вряд ли что-то получится. У меня было несколько заказчиков, которые потерпели крах, но смогли привести все в порядок с нашей помощью.

Скорее всего, у вас не возникнет проблем с количеством партиций и масштабируемостью, даже пока Kafka еще использует ZooKeeper, если вы правильно все спроектируете и разберетесь в базовых принципах Kafka. Такие проблемы часто возникают с любой технологией, вроде Kafka, которая вводит совершенно новый уровень и парадигму (например, при первом появлении облаков немало компаний набили себе шишки при попытке их освоить).

Связанный миф: Pulsar поддерживает практически неограниченное число партиций.

Неправда.

У BookKeeper существуют те же ограничения по одному файлу на ledger, что и у Kafka, но в одной партиции таких ledger несколько. Брокеры Pulsar объединяют партиции в группы, но на уровне хранения, в Bookkeeper, данные хранятся в сегментах, и на каждую партицию приходится много сегментов.

В Kafka метаданные для этих сегментов хранятся в Zookeeper, откуда и возникают эти ограничения по числу. Когда Kafka избавится от этой зависимости, границы возможного раздвинутся. С нетерпением жду реализации KIP-500. Подробности читайте в статье Apache Kafka справится сама: удаление зависимости от Apache ZooKeeper.

Связанный миф: потребности масштабирования в Kafka необходимо определять при создании топика.

Отчасти это правда.

Если требуется очень большой масштаб, в топиках Kafka можно сразу создать больше партиций, чем обычно требуется для этой задачи. (См. Потоки и таблицы в Apache Kafka: топики, партиции и хранение.) Или можно добавить партиции позже. Это не идеальное решение, но так уж устроены распределенные системы стриминга (которые, кстати, масштабируются лучше, чем традиционные системы обработки сообщений, вроде IBM MQ).

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

Но! У топиков Pulsar есть то же ограничение!

Пропускная способность для записи зависит от числа партиций, назначенных топику Pulsar, точно так же, как это происходит в топике Kafka, поэтому там тоже придется создавать партиции с запасом. Дело в том, что в каждой партиции записывать можно только в один ledger за раз (а их может быть несколько). Увеличение числа партиций динамически влияет на порядок сообщений, как и в Kafka (порядок нарушается).

У Kafka и Pulsar с масштабированием все отлично. Этого будет более чем достаточно почти в любом сценарии.

Если вам нужен совсем заоблачный масштаб, лучше взять реализацию без ZooKeeper. Так что KIP-500 это самое ожидаемое изменение в Kafka, судя по сообществу и заказчикам Confluent.

Миф 10. В случае аппаратного сбоя Pulsar восстанавливается моментально, а Kafka приходится перегружать данные.

И да, и нет.

Если брокер Pulsar вырубится, ничего не страшного не будет, это правда, но, в отличие от брокера Kafka, он и не хранит данные, а просто выступает в качестве прокси для уровня хранения, то есть BookKeeper. Так что подобные заявления о Pulsar это просто маркетинговый ход. Почему никто не говорит, что бывает, если полетит нода BookKeeper (bookie)?

Если завершить и перезапустить BookKeeper bookie, данные будут точно так же перераспределяться, как и в Kafka. В этом суть распределенных систем с их репликацией и партициями.

Kafka уже предлагает эластичность.

Это важно. Основатель Confluent, Джей Крепс (Jay Kreps) недавно писал об этом: Эластичные кластеры Apache Kafka в Confluent Cloud. В облачном сервисе SaaS, вроде Confluent Cloud, конечный пользователь не думает об аппаратных сбоях. Он ожидает непрерывный аптайм и SLA на уровне 99,xx. С оплатой за потребление пользователь не заботится об управлении брокерами, изменении размеров нод, увеличении или уменьшении кластеров и подобных деталях.

Самоуправляемым кластерам Kafka нужны такие же возможности. Tiered Storage for Kafka это огромное хранилище, с которым данные не хранятся на брокере и восстановление после сбоев происходит почти моментально. Если добавить сюда такие инструменты, как Self-Balancing Kafka (эта фича от Confluent описана в статье по ссылке выше), можно вообще забыть об эластичности в самоуправляемых кластерах.

К сожалению, в Pulsar вы ничего подобного не найдете.

Миф 11. Pulsar справляется с репликацией между кластерами лучше, чем Kafka.

Неправда.

Любая распределенная система должна решать такие проблемы, как CAP-теорема и кворум в распределенных вычислениях. Кворум это минимальное число голосов, которое распределенная транзакция должна получить, чтобы выполнить операцию в распределенной системе. Такой подход позволяет обеспечить согласованность операций.

Задачи по кворуму Kafka поручает ZooKeeper. Даже после реализации KIP-500 и удаления ZooKeeper законы физики не перестанут действовать: проблемы задержки в распределенных системах существуют в таких регионах, как Восточная, Центральная и Западная часть США и даже по всему миру. Скорость света, конечно, впечатляет, но все же имеет ограничения.

Эту проблему можно обойти разными способами, включая использование инструментов репликации в реальном времени, например MirrorMaker 2 в Apache Kafka, Confluent Replicator или Confluent Multi-Region-Clusters. Эти варианты и советы см. в статье Архитектурные шаблоны для распределенных, гибридных, периферийных и глобальных деплоев Apache Kafka.

Вы не найдете универсальное решение, которое обеспечит глобальную репликацию + нулевой простой + нулевую потерю данных. Для самых критически важных приложений Confluent Multi-Region-Clusters предлагает RTO=0 и RPO=0 (нулевой простой и нулевую потерю данных) с автоматическим аварийным восстановлением и отработкой отказа, даже если упадет целый датацентр или облачный регион.

Для таких случаев архитектура Pulsar будет еще сложнее, чем ее базовая версия. Просто для георепликации Pulsar требует дополнительный глобальный кластер Zookeeper, а для больших расстояний это не годится. Обходной путь есть, но от CAP-теоремы и законов физики не уйдешь.

Что с Kafka, что с Pulsar вам нужна проверенная архитектура для борьбы с физикой в глобальных деплоях.

Миф 12. Pulsar совместим с интерфейсом и API Kafka.

Отчасти.

Pulsar предлагает простейшую реализацию с зачаточной совместимостью с протоколом Kafka v2.0.

У Pulsar есть конвертер для базовых элементов протокола Kafka.

В теории совместимость с Kafka выглядит убедительно, но вряд ли это серьезный аргумент для переноса действующей инфраструктуры Kafka в Pulsar. Зачем такой риск?

Мы слышали заявления о совместимости с Kafka и в других примерах, например для гораздо более зрелого сервиса Azure Event Hubs. Почитайте об ограничивающих факторах их Kafka API. Вы удивитесь. Никакой поддержки основных функций Kafka, вроде транзакций (и семантики exactly-once), сжатия или compaction для логов.

Раз, по сути, это не Kafka, существующие приложения Kafka будут вести себя непредсказуемо в таких вот совместимых системах, будь то Azure Event Hubs, Pulsar или другая оболочка.

Kafka vs. Pulsar комплексное сравнение

Выше мы говорили о мифах, которые гуляют по разным статьям и постам. С ними, вроде, все понятно.

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

Рассмотрим три из них: доля на рынке, корпоративная поддержка и облачные предложения.

Доля на рынке у Apache Kafka и Apache Pulsar

Статистика в Google Trends за последние пять лет совпадает с моими личными наблюдениями интерес к Apache Kafka гораздо выше, чем к Apache Pulsar:

Примерно так же выглядит картинка сравнения на Stack Overflow и аналогичных платформах, по числу и размеру поддерживающих вендоров, открытой экосистеме (интеграция инструментов, обертки, вроде Spring Kafka) и подобным характеристикам.

Открытые вакансии еще один индикатор распространения технологии. Для Pulsar их мало, то есть его использует не так много компаний. Убедитесь сами поищите на любом сайте. Если искать по всему миру, для Pulsar вакансий меньше сотни, а для Kafka тысячи. Кроме того, в большинстве вакансий для Pulsar указано, что требуется опыт с Kafka, Pulsar, Kinesis и аналогичными технологиями.

В большинстве случаев такие характеристики важнее для успешного проекта, чем технические детали. В конце концов ваша цель решить определенную бизнес-задачу, правда же?

Раз Pulsar так мало распространен, почему о нем вообще говорят? Во-первых, потому, что независимые консультанты, аналитики и блоггеры (включая меня) должны рассказывать о новых современных технологиях, чтобы привлечь аудиторию. Если честно, люди любят об этом читать.

Корпоративная поддержка для Kafka и Pulsar

Корпоративная поддержка для Kafka и Pulsar существует.

Но все не совсем так, как можно ожидать. Вот несколько вендоров, к которым вы можете обратиться, если решили работать с Pulsar:

• Streamlio (теперь принадлежит Splunk) компания, которая раньше стояла за Apache Pulsar. Splunk пока не объявили о своей будущей стратегии для поддержки пользователей, которые работают над проектами на основе Pulsar. Splunk славится своими популярными аналитическими платформами. Это их основной бизнес (1,8 млрд долларов в 2019 году). Единственное, на что жалуются пользователи, ценник. Splunk активно используют Kafka, а сейчас интегрируют Pulsar в Splunk Data Stream Processor (DSP). Сильно сомневаюсь, что Splunk вдруг увлечется опенсорсом, чтобы поддержать ваш драгоценный проект на базе Pulsar (возможно, чего-то можно ожидать от DSP). Будущее покажет.
• StreamNative основана одним из изначальных разработчиков Apache Pulsar и поддерживает платформу стриминга на основе Pulsar. На июнь 2020 года у StreamNative было 13 (целых 13, да!) сотрудников в LinkedIn. Уж не знаю, хватит этого для поддержки вашего критически важного проекта или нет.
• TIBCO объявили о поддержке Pulsar в декабре 2019 года. За последние годы стратегия у них сместилась с интеграции на аналитику. Пользователи их промежуточного ПО уходят от TIBCO толпами. В отчаянии они приняли стратегическое решение: поддерживать другие платформы, даже если у них нет никакого опыта и никакой связи с ними. Да, звучит как миф, но TIBCO делают то же самое для Kafka. Любопытный факт: TIBCO предлагает Kafka и ZooKeeper в Windows! Такого больше никто не делает. Все знают, это приводит к нестабильности и несогласованности. Но если что, TIBCO помогут вам с Kafka и Pulsar. Зачем вообще сравнивать эти платформы, если можно получить обе у одного вендора, причем даже на Windows, с .exe и bat-скриптами для запуска серверных компонентов:

Вендоров для Kafka больше с каждым днем.

Это очень распространенная на рынке платформа. Лучшее подтверждение поддержка от крупнейших вендоров. IBM, Oracle, Amazon, Microsoft и многие другие поддерживают Kafka и встраивают возможности интеграции с ней в свои продукты.

Окончательное подтверждение этому я получил на конференции Oracle OpenWorld 2019 в Сан-Франциско, где выступал менеджер по GoldenGate (великолепный и очень дорогой инструмент CDC от Oracle). В основном он рассказывал, как GoldenGate станет платформой интеграции данных вообще для всего. Половина выступления была посвящена стримингу, платформе Kafka и тому, что GoldenGate будет обеспечивать интеграцию с разными базами/озерами данных и Kafka в обоих направлениях.

Полностью управляемые облачные сервисы для Kafka и Pulsar

Какие облачные решения доступны для Kafka и Pulsar?

Для Apache Pulsar есть облачный сервис с очень говорящим названием:

Kafkaesque.

Я серьезно. Сами посмотрите [UPD: 17 июня они переименовали KAFKAESQUE в KESQUE. Видимо, поняли, всю комичность ситуации.]

Для Apache Kafka существуют разные предложения на выбор:

• Confluent Cloud (SaaS) полностью управляемый сервис с оплатой по мере использования, круглосуточной доступностью, эластичностью и бессерверными функциями для Apache Kafka и связанной экосистемы (Schema Registry, коннекторы Kafka Connect и ksqlDB для обработки потоков).
• Amazon MSK (PaaS) подготавливает ZooKeeper и брокеры Kafka, а конечные пользователи сами обслуживают их, исправляют баги, устанавливают обновления и т. д. Важный факт: AWS не включает проблемы с Kafka в поддержку и SLA 99,95.
• Azure Event Hubs (SaaS) предоставляет конечную точку Kafka (с проприетарной реализацией) для взаимодействия с приложениями Kafka. Это легко масштабируемое решение с высокой производительностью. Это не вполне Kafka, просто эмуляция, и тут не хватает некоторых важных функций, вроде семантики exactly-once, compaction логов и сжатия. Не говоря уже о дополнительных возможностях, вроде Kafka Connect и Kafka Streams
• Big Blue (IBM) и Big Red (Oracle) предлагают облачные решения для Kafka и соответствующих API. Не знаю, использует их кто-нибудь или нет. Понятия не имею, насколько они хороши, никогда не видел их в деле.
• Много компаний поменьше, вроде Aiven, CloudKarafka, Instaclustr и других.

В общем, про распространенность Kafka и Pulsar на рынке все очевидно.

И все же Apache Kafka или Apache Pulsar?

В двух словах: Pulsar пока здорово отстает от Kafka по зрелости в плане надежности для больших масштабов и обширности сообщества.

И вообще не факт, что Pulsar лучше.

Есть смысл сравнивать их, если вы решили использовать только опенсорс-инструменты. Узнайте, что подойдет лучше именно вам. Не забудьте учесть технические характеристики, уровень зрелости, вендоров, сообщество разработчиков и другие важные факторы. Что вписывается в вашу ситуацию?

Если в Kafka и Pulsar есть не все нужные вам функции, выбирайте Kafka: для нее есть много вендоров и облачных решений. Pulsar, к сожалению, сильно отстает и вряд ли догонит в ближайшем будущем.

В основе Apache Kafka находится лог простая структура данных, которая использует последовательные операции, работающие в симбиозе с оборудованием. Эффективное использование дискового буфера и кэша процессора, prefetch, передача данных zero-copy и много других радостей все это благодаря построенной на логе структуре, которая славится своей эффективностью и пропускной способностью. Обычно эти преимущества, а еще базовая реализация в виде лога коммитов, первое, что люди узнают о Kafka.

Код самого лога составляет относительно малую часть всей системы. Гораздо больше занимает код, который отвечает за организацию партиций (т. е. логов) на множестве брокеров в кластере назначает лидеров, обрабатывает сбои и т. д. Этот код и делает Kafka надежной распределенной системой.

Раньше важной частью работы распределенного кода был Apache ZooKeeper. Он хранил самые важные метаданные системы: где находятся партиции, кто из реплик лидер и т. д. Поначалу эффективный и проверенный ZooKeeper действительно был нужен, но, по сути, это что-то вроде специфической файловой системы или API триггеров поверх согласованного лога. А Kafka это API pub/sub (издатель-подписчик) поверх согласованного лога. В результате пользователи настраивают, конфигурируют, мониторят, защищают и обдумывают взаимодействие и производительность между двумя реализациями лога, двумя сетевыми уровнями и двумя реализациями системы безопасности каждая со своими инструментами и хуками мониторинга. Все стало неоправданно сложно, поэтому решено было заменить ZooKeeper сервисом кворума прямо в самой Kafka.

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

Мы с Джейсоном, Колином и командой KIP-500 прогнали полный жизненный цикл сервера Kafka с созданием и потреблением сообщений и все без Zookeeper. Какая же красота!
Бен Стопфорд (Ben Stopford)

Рады сообщить, что ранний доступ к KIP-500 уже закоммичен в ствол и будет включен в предстоящий релиз 2.8. Впервые Kafka можно использовать без ZooKeeper. Мы называем это режимом Kafka Raft Metadata, сокращенно KRaft (читается крафт).

В этом релизе пока доступны не все фичи. Мы еще не поддерживаем ACL и другие транзакции или функции безопасности. Переназначение партиций и JBOD тоже не поддерживаются в режиме KRaft (надеемся, что они будут доступны в одном из релизов Apache Kafka в этом году). Так что контроллер кворума считаем экспериментальным и для прода пока не используем. В остальном преимуществ масса: деплоймент и обслуживание станут проще, всю Kafka можно запустить как один процесс, а на кластер помещается гораздо больше партиций (цифры см. ниже).

Контроллер кворума: консенсус на основе событий

Если запустить Kafka с новым контроллеров кворума, все задачи по метаданным, которые раньше выполнялись контроллером Kafka и ZooKeeper, ложатся на новый сервис, запущенный прямо в кластере Kafka. Контроллер кворума можно запустить на выделенном оборудовании, если этого требует ваш сценарий.

Внутри происходит много интересного. Контроллеры кворума используют новый протокол KRaft, чтобы точно реплицировать метаданные по всему кворуму. Протокол во многом напоминает ZooKeeper ZAB и Raft, но с важными отличиями например, что логично, он использует архитектуру на основе событий.

Контроллер кворума хранит свой стейт с помощью модели на основе событий, чтобы стейт-машину всегда можно было в точности воссоздать. Чтобы лог события, который хранит стейт (также называется топиком метаданных), не разрастался до бесконечности, периодически он обрезается с созданием снапшотов. Остальные контроллеры в кворуме фолловят активный контроллер и реагируют на события, которые он создает и сохраняет в свой лог. Если одна нода приостановлена во время разделения на партиции, например, она быстро догонит пропущенные события с помощью лога. Так мы существенно сокращаем период простоя и время восстановления системы в худшем сценарии.

Консенсус на основе событий

Раз протокол KRaft управляется событиями, в отличие от контроллера ZooKeeper контроллер кворума не загружает стейт из ZooKeeper, когда становится активным. При смене лидера у нового активного контроллера уже есть в памяти все закоммиченные записи метаданных. Более того, для отслеживания метаданных в кластере используется тот же механизм на основе событий, что и в протоколе KRaft. Задача, которую раньше обрабатывали RPC, теперь полагается на события и использует лог для коммуникации. Приятное последствие этих изменений (и изменений, которые были предусмотрены изначальным замыслом) Kafka теперь поддерживает гораздо больше партиций, чем раньше. Обсудим подробнее.

Увеличение масштаба Kafka до миллионов партиций

Максимальное число партиций в кластере Kafka определяется двумя свойствами: лимит партиций для ноды и лимит партиций для кластера. Сейчас на ограничение в кластере во многом влияет управление метаданными. Предыдущие KIP улучшили лимит для ноды, хотя и здесь нет предела совершенству. Масштабируемость в Kafka зависит от добавления нод. Здесь нам важен лимит для кластера, который определяет верхнюю границу масштабирования всей системы.

Новый контроллер кворума может обрабатывать гораздо больше партиций на кластере. Чтобы это проверить, можно провести такие же тесты, как в 2018 году, с помощью которых мы демонстрировали лимиты партиций в Kafka. Эти тесты измеряют время, которое требуется для завершения работы и восстановления. В старом контроллере это операция O (число партиций). Она устанавливает верхнюю границу числа партиций, которые сейчас поддерживает один кластер Kafka.

Предыдущая реализация, как рассказывалось в посте по ссылке, могла достигать 200K партиций. Ограничивающим фактором служило время, которое требовалось для переноса важных метаданных между внешним консенсусом (ZooKeeper) и внутренним управлением лидером (контроллер Kafka). С новым контроллером кворума обе роли выполняет один компонент. Подход на основе событий означает, что контроллер отрабатывает отказ почти моментально. Ниже приводятся цифры нашего тестирования для кластера с 2 млн партиций (в 10 раз больше предыдущего лимита):

Важно учитывать контролируемую и неконтролируемую остановку. Контролируемые остановки включают типичные рабочие сценарии, вроде rolling restart стандартная процедура развертывания изменений без потери доступности. Восстановление после неконтролируемой остановки важнее, потому что оно задает целевое время восстановления (RTO) системы в случае, скажем, неожиданного сбоя, например, на виртуальной машине или поде, или потери связи с дата-центром. Хотя эти цифры указывают на производительность всей системы, они напрямую определяют хорошо известное узкое место, которое возникает при использовании ZooKeeper.

Измерения для контролируемой и неконтролируемой остановки не сравниваются напрямую, потому что неконтролируемая остановка включает выбор новых лидеров, а контролируемая нет. Это расхождение введено специально, чтобы контролируемая остановка соответствовала изначальным тестам 2018 года.

Упрощение Kafka единый процесс

Kafka часто воспринимается как тяжеловес, и такой репутацией она во многом обязана сложностями с управлением ZooKeeper. Из-за этого для запуска проектов часто выбирают более легкую очередь сообщений, вроде ActiveMQ или RabbitMQ, а Kafka появляется на сцене только при разрастании масштаба.

Это печально, ведь Kafka дает удобную абстракцию на основе лога коммитов, которая одинаково хорошо подходит для маленьких рабочих нагрузок у стартапов и для бешеного трафика у Netflix или Instagram. А если вам нужна стриминговая обработка, без Kafka с абстракцией лога коммитов не обойтись, будь то Kafka Streams, ksqlDB или другой аналогичный фреймворк. Управлять двумя отдельными системами (Kafka и Zookeeper) сложно, поэтому пользователи выбирали между масштабированием и простотой.

Теперь можно получить все и сразу. Благодаря KIP-500 и режиму KRaft мы можем легко начать проект с Kafka или использовать его как альтернативу монолитным брокерам, вроде ActiveMQ или RabbitMQ. Благодаря легкости и единому процессу это подходящий вариант для ограниченных в ресурсах устройств. В облаке все еще проще. Управляемые сервисы, например, Confluent Cloud, все берут на себя. Не важно, собственный это будет кластер или управляемый, мы можем начать с малого, а потом расширять его вместе со своими рабочими нагрузками все в рамках одной инфраструктуры. Давайте посмотрим, как выглядит развертывание с одним процессом.

Испытываем Kafka без ZooKeeper

Новый контроллер кворума пока доступен в экспериментальном режиме, но, скорее всего, будет включен в релиз Apache Kafka 2.8. На что он способен? Самое крутое это, конечно, возможность создавать кластер Kafka с одним процессом, как в демо.

Конечно, если вам нужна очень высокая пропускная способность, и вы хотите добавить репликацию для отказоустойчивости, понадобятся еще процессы. Пока контроллер кворума на базе KRaft доступен только как Early Access, для критических рабочих нагрузок он не годится. В ближайшие месяцы мы будем добавлять недостающие фрагменты, моделировать протокол с помощью TLA+ и внедрять контроллер кворума в Confluent Cloud.

Но поэкспериментировать вы можете уже сейчас. См. полный README на GitHub.

Гостем подкаста The Art Of Programming стал спикер курса Слёрма по Kafka Александр Миронов, Infrastructure Engineer в Stripe. Тема выпуска Проблемная Kafka. Обсудили вопросы, часто возникающие при работе с Kafka: аудит входных данных, квоты, способы хранения данных, возможный даунтайм в консьюмер-группах и др.

Итоговый проект по Kafka

Релиз видеокурса от Александра Миронова и Анатолия Солдатова состоялся 7 апреля. Кроме заявленных тем мы добавили в курс итоговый проект: студентам нужно будет выполнить практическое задание под звёздочкой с применением знаний, полученных на курсе.

Мы добавили два варианта заданий из области разработки и из области администрирования. В первом варианте понадобится создать приложение для генерации real-time статистики продаж интернет-магазина, а во втором спасти кластер от хаоса.

Подробнее о курсе
Бесплатные материалы курса

Все больше инженеров выбирают Kotlin для разработки серверных приложений. Полная совместимость с Java, корутины и высокая безопасность делают Kotlin отличным инструментом для подобных задач.

Мы организуем серию вебинаров (на английском языке), где расскажем о разработке бэкенда на Kotlin в сочетании с технологиями Apache Kafka, Spring Boot и Google Cloud Platform. Вебинары подойдут для Kotlin- и Java-разработчиков любого уровня подготовленности, в том числе для разработчиков мобильных приложений без опыта серверной разработки.

Kotlin и Apache Kafka, 10 декабря 2020, 19:30 20:30 МСК
Kotlin и Google Cloud Platform, 17 декабря 2020, 19:30 20:30 МСК
Kotlin и Spring Boot, 14 января 2021, 19:30 20:30 МСК

Подробнее о каждом из вебинаров читайте ниже.

Kotlin и Apache Kafka

Зарегистрироваться

O чем этот вебинар?

Виктор и Антон покажут, как использовать Apache Kafka в сочетании с Kotlin для управления потоками данных. Презентация также даст обзор внутренней архитектуры Apache Kafka.

Спикеры:

• Антон Архипов, Developer Advocate в команде Kotlin, JetBrains
• Виктор Гамов, Developer Advocate, Confluent

Когда состоится вебинар?

10 декабря 2020
19:30 20:30 МСК
Подробная информация

Kotlin и Google Cloud Platform

Зарегистрироваться

О чем этот вебинар?

На вебинаре мы расскажем, как применить Kotlin в проекте с кодовой базой на Java, поговорим о таких возможностях Kotlin, как корутины, и продемонстрируем процесс развертывания приложения на Google Cloud.

Спикер: Джеймс Уорд, Developer Advocate, Google Cloud Platform

Когда состоится вебинар?

17 декабря 2020
19:30 20:30 МСК
Подробная информация

Kotlin и Spring Boot

Зарегистрироваться

О чем этот вебинар?

В ходе вебинара Рэй поможет вам создать бэкенд-сервис с использованием Spring Boot и Spring Cloud GCP и покажет полезные сервисы Google Cloud для баз данных, хранения и мониторинга состояния. Когда приложение будет готово, Рэй покажет, как с помощью Google Cloud Platform сделать из него бессерверный сервис, и продемонстрирует возможности автоматического масштабирования.

Спикер: Рэй Цанг, Developer Advocate в Google Cloud Platform и Java Champion

Когда состоится вебинар?

14 января 2021
19:30 20:30 МСК
Подробная информация

После регистрации вы получите письмо с подтверждением и приглашением, которое сможете добавить в календарь. Мы пришлем вам напоминание за день до начала вебинара.

Все вебинары бесплатные. Записи будут доступны на канале JetBrains TV в Youtube. Подпишитесь на JetBrains TV, чтобы не пропустить появление записей.

Узнайте больше о бэкенд-разработке на Kotlin

Мы создали страницу, посвященную разработке серверных приложений на Kotlin.
Здесь вы найдете список фреймворков и библиотек, учебные материалы, примеры использования и многое другое.

Ваша команда Kotlin
The Drive to Develop

Введение

Для анализа потоковых данных необходимы источники этих данных. Так же важна сама информация, которая предоставляется источниками. А источники с текстовой информацией, к примеру, еще и редки.
Из интересных источников можно выделить следующие: twitter, vk. Но эти источники подходят не под все задачи.
Есть источники с нужными данными, но эти источники не потоковые. Здесь можно привести следующее ссылки: public-apis.
При решении задач, связанных с потоковыми данными, можно воспользоваться старым способом.
Скачать данные и отправить в поток.
Для примера можно воспользоваться следующим источником: imdb.
Следует отметить, что imdb предоставляет данные самостоятельно. См. IMDb Datasets. Но можно принять, что данные собранные напрямую содержат более актуальную информацию.

Язык: Java 1.8.
Библиотеки: kafka 2.6.0, jsoup 1.13.1.

Сбор данных

Сбор данных представляет из себя сервис, который по входным данным загружает html-страницы, ищет нужную информацию и преобразует в набор объектов.
Итак источник данных: imdb. Информация будет собираться о фильмах и будет использован следующий запрос: https://www.imdb.com/search/title/?release_date=%s,%s&countries=%s
Где 1, 2 параметр это даты. 3 параметр страны.
Для лучшего понимания источника данных можно обратится к следующему ресурсу: imdb-extensive-dataset.

Интерфейс для сервиса:

public interface MovieDirectScrapingService {    Collection<Movie> scrap();}

Класс Movie это класс, которые содержит информацию об одном фильме (или о шоу и т.п.).

class Movie {    public final String titleId;    public final String titleUrl;    public final String title;    public final String description;    public final Double rating;    public final String genres;    public final String runtime;    public final String baseUrl;    public final String baseNameUrl;    public final String baseTitleUrl;    public final String participantIds;    public final String participantNames;    public final String directorIds;    public final String directorNames;

Анализ данных на одной странице.
Информация собирается следующим образом. Данные закачиваются с помощью jsoup. Далее ищутся нужные html-элементы и трансформируются в экземпляры для фильмов.

String scrap(String url, List<Movie> items) {    Document doc = null;    try {        doc = Jsoup.connect(url).header("Accept-Language", language).get();    } catch (IOException e) {        e.printStackTrace();    }    if (doc != null) {        collectItems(doc, items);        return nextUrl(doc);    }    return "";}

Поиск ссылки на следующею страницу.

String nextUrl(Document doc) {    Elements nextPageElements = doc.select(".next-page");    if (nextPageElements.size() > 0) {        Element hrefElement = nextPageElements.get(0);        return baseUrl + hrefElement.attributes().get("href");    }    return "";}

Тогда основной метод будет таким. Формируется начальная строка поиска. Закачиваются данные по одной странице. Если есть следующая страница, то идет переход к ней. По окончании передаются накопленные данные.

@Overridepublic Collection<Movie> scrap() {    String url = String.format(            baseUrl + "/search/title/?release_date=%s,%s&countries=%s",            startDate, endDate, countries    );    List<Movie> items = new ArrayList<>();    String nextUrl = url;    while (true) {        nextUrl = scrap(nextUrl, items);        if ("".equals(nextUrl)) {            break;        }        try {            Thread.sleep(50);        } catch (InterruptedException e) {        }    }    return items;}

Подробности по остальным методам можно найти в ссылках на ресурсы.

Отправка данных в топик

Формируется следующий сервис: MovieProducer. Здесь будет один единственный публичный метод: run.

Создается продюсер для кафки. Загружаются данные из источника. Трансформируются и отправляются в топик.

public void run() {    try (SimpleStringStringProducer producer = new SimpleStringStringProducer(            bootstrapServers, clientId, topic)) {        Collection<Data.Movie> movies = movieDirectScrapingService.scrap();        List<SimpleStringStringProducer.KeyValueStringString> kvList = new ArrayList<>();        for (Data.Movie move : movies) {            Map<String, String> map = new HashMap<>();            map.put("title_id", move.titleId);            map.put("title_url", move.titleUrl);                        String value = JSONObject.toJSONString(map);            String key = UUID.randomUUID().toString();            kvList.add(new SimpleStringStringProducer.KeyValueStringString(key, value));        }        producer.produce(kvList);    }}

Теперь все вместе

Формируются нужные параметры для поиска. Загружаются данные и отправляются в топик.
Для этого понадобится еще один класс: MovieDirectScrapingExecutor. С одним публичным методом: run.

В цикле создаются данные для поиска из текущей даты. Происходит загрузка и отправка данных в топик.

public void run() {    int countriesCounter = 0;    List<String> countriesSource = Arrays.asList("us");    while (true) {        try {            LocalDate localDate = LocalDate.now();            int year = localDate.getYear();            int month = localDate.getMonthValue();            int day = localDate.getDayOfMonth();            String monthString = month < 9 ? "0" + month : Integer.toString(month);            String dayString = day < 9 ? "0" + day : Integer.toString(day);            String startDate = year + "-" + monthString + "-" + dayString;            String endDate = startDate;            String language = "en";            String countries = countriesSource.get(countriesCounter);            execute(language, startDate, endDate, countries);            Thread.sleep(1000);            countriesCounter += 1;            if (countriesCounter >= countriesSource.size()) {                countriesCounter = 0;            }        } catch (InterruptedException e) {        }    }}

Для запуска потребуется экземпляр класса MovieDirectScrapingExecutor, который можно запустить с нужными параметрами, к примеру, из метода main.

Пример отправляемых данных для одного фильма.

{  "base_name_url": "https:\/\/www.imdb.com\/name",  "participant_ids": "nm7947173~nm2373827~nm0005288~nm0942193~",  "title_id": "tt13121702",  "rating": "0.0",  "base_url": "https:\/\/www.imdb.com",  "description": "It's Christmas time and Jackie (Carly Hughes), an up-and-coming journalist, finds that her life is at a crossroads until she finds an unexpected opportunity - to run a small-town newspaper ... See full summary ",  "runtime": "",  "title": "The Christmas Edition",  "director_ids": "nm0838289~",  "title_url": "\/title\/tt13121702\/?ref_=adv_li_tt",  "director_names": "Peter Sullivan~",  "genres": "Drama, Romance",  "base_title_url": "https:\/\/www.imdb.com\/title",  "participant_names": "Carly Hughes~Rob Mayes~Marie Osmond~Aloma Wright~"}

Подробности можно найти в ссылках на ресурсы.

Тесты

Для тестирования основной логики, которая связана с отправкой данных, можно воспользоваться юнит-тестами. В тестах предварительно создается kafka-сервер.
См. Apache Kafka и тестирование с Kafka Server.

Сам тест: MovieProducerTest.

public class MovieProducerTest {    @Test    void simple() throws InterruptedException {        String brokerHost = "127.0.0.1";        int brokerPort = 29092;        String zooKeeperHost = "127.0.0.1";        int zooKeeperPort = 22183;        String bootstrapServers = brokerHost + ":" + brokerPort;        String topic = "q-data";        String clientId = "simple";        try (KafkaServerService kafkaServerService = new KafkaServerService(                brokerHost, brokerPort, zooKeeperHost, zooKeeperPort        )        ) {            kafkaServerService.start();            kafkaServerService.createTopic(topic);            MovieDirectScrapingService movieDirectScrapingServiceImpl = () -> Collections.singleton(                    new Data.Movie()            );            MovieProducer movieProducer =                    new MovieProducer(bootstrapServers, clientId, topic, movieDirectScrapingServiceImpl);            movieProducer.run();            kafkaServerService.poll(topic, "simple", 1, 5, (records) -> {                assertTrue(records.count() > 0);                ConsumerRecord<String, String> record = records.iterator().next();                JSONParser jsonParser = new JSONParser();                JSONObject jsonObject = null;                try {                    jsonObject = (JSONObject) jsonParser.parse(record.value());                } catch (ParseException e) {                    e.printStackTrace();                }                assertNotNull(jsonObject);                    });            Thread.sleep(5000);        }    }}

Заключение

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

Ссылки и ресурсы

Исходный код.

Я учусь иллюстрировать сложные процессы с помощью комиксов. Нашла себе в копилку крутой кейс: как с помощью комиксов про милых выдр можно ребенку объяснить такую сложную штуку как Apache Kafka, и сделать мир немного добрее.

Легко по течению легкое введение в потоковую обработку и Apache Kafka. Группа выдр обнаруживает, что они могут использовать гигантскую реку для общения друг с другом. По мере того, как все больше выдр перемещается в лес, они должны научиться адаптировать свою систему, чтобы справиться с возросшей активностью леса.

Под катом 25 слайдов, объясняющие основы Kafka для детей и гуманитариев. И много милых выдр.

Легко по течению

Поначалу, в лесу было тихо, и только две семьи выдр жили среди деревьев и рек.



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

Они делились новостями о вечеринка в честь дня рождения, новых гостях и других событиях, происходящих в лесу.



Со временем, все больше и больше выдр переселялось в лес.

И точно так же росло количество происходящих в лесу событий.



Это было целым вызовом для маленьких выдр. Каждый раз, когда у выдры были новости, чтобы поделиться ими, она должна была найти каждую другую семью и рассказать им лично.



Это не только занимало много времени, но было чревато ошибками.

Что если какая-то выдрячья семья была на пикнике, и не могла получить уведомление о предстоящем событии?



Ну, тогда выдре, которая хотела оповестить соседей, придется вернуться позже или сдаться.



Поскольку остальные выдры общались друг с другом лично, говорили, что они Тесно Связаны.

Быть тесно связанными супер мило, но это делает коммуникацию с другими сложнее. Это называется Проблема Масштабирования.


Именно тогда у одной маленькой выдры по имени Никси, появилась идея, которая навсегда изменила лес.

По лесу текла огромная река Кафка, и Никси знала, что выдры могу использовать эту реку для общения.

Вы можете установить Kafka в своем лесу:

# clone the repositorygit clone \https://github.com/round-robin-books/gently-kafka.git# start kafkadocker-compose up

Она даже сочинила песню, чтобы объяснить, как это работает:

События свои
В реку ты опусти,
Река их отнести
Сможет выдрам другим.

Что о событиях,
Плывущих в потоке,
Этот путь удивителен,
Мы все тут в шоке.



Не отстанешь ли ты? вмешался дельфин
Нельзя оставлять все на милость глубин!

Не будет в доставке такого.
Мы командой реку разделим,
На множество мелких потоков
И вместе достигнем цели.

Давайте посмотрим, как это рабоnает

Во первых, выдра наблюдает Событие, что-то, что произошло в определенный момент времени.

К примеру, Сегодня вернулись пчелы это событие.



Дальше, выдра создает Запись о событии.
Записи (которые иногда называют сообщения) включают время события (отметка о времени) и дополнительную информацию о событии.



Тогда выдры смогут решать в какую часть реки направить это сообщение.

Река делится на потоки, которые называют Топики, которые делают организацию сообщений проще.



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

По началу они использовали стеклянные бутылки, но бутылки просто уплывали.



Тогда, выдры переключились на стеклянные поплавки. Поскольку поплавки оставались на месте и никуда не уплывали, это было названо Устойчивостью.

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

2 типа выдр

Выдры, которые помещают сообщения в реку, называются Продюсеры.

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

Незнание помогает Разделять системы, то есть выдр, которые создают события, и которые читают события.



Выдры, которые читают события в потоке называются Консьюмеры.

Консьюмеры следят за любым топиком, который их интересует.

Например, выдры Первой помощи следят за топиком с лесными алертами, так что они могут реагировать в чрезвычайных ситуациях (как в ситуации с пчелами).



Эта система работает хорошо, но спустя какое-то время река становится слишком загруженной огромным количеством событий. Как же маленьким выдрам все успеть?

Будучи социальными животными, они предпочитают работать вместе. Во первых, они скидывают огромные камни в реку, разделяя каждый топик на небольшое количество потоков, или Разделов.



Тогда, один член семьи, так называемый Лидер Группы, назначает подмножество мелких потоков (разделов) каждому члену семьи.

Это подразумевает, что каждая выдра ответственна только за ту часть потока, которую она мониторит.



Работая совместно в этом случае, каждая выдрячья семья называется Группой Консьюмеров, которая успевает следить за своей частью потока сообщений, сколько бы их не было всего в реке.



Более того, если выдра заболела или у нее есть дела, ее работу можно поручить другой выдре в ее группе.

Поскольку кто-то всегда был рядом, выдры были Высоко доступными.

И поскольку они могли справляться с Незапланированными Ситуациями, они были Отказоустойчивыми.



Еще существовала магическая часть леса, Земля Потоковой Обработки, где выдры могли делать реально классные вещи с событиями в реке.



Целая книга написана об этом магическом месте: Mastering Kafka Streams and ksqlDB



Выдры также построили дороги, которые называются Кафка Коннекторы, чтобы донести сообщения других сообществ в реку Кафка. Все присоединились к веселью.



Кафка продолжала помогать многим другим во всем мире. И возвращалась обратно в лес.

Жизнь шла своим чередом многие годы, а выдры жили долго и счастливо.

---

Я не удержалась, вот вам несколько фактов про выдр:



Выдрята, или щенки выдр, спят у мам-выдр на груди.



Взрослые выдры держат друг друга за лапки во сне, чтобы их не разделило течение.



А одинокие выдры, чтобы не дрейфовать, заворачиваются в водоросли.

Строго говоря, автор использует термин otter, что переводится как выдровые, и включает себя 13 видов животных с распространением по всему миру, кроме Австралии.

И если эти слайды вы действительно использовали этот материал на детях в образовательных целях, то завершить можно речевым упражнением:

В недрах тундры выдры в гетрах тырят в ведра ядра кедров. Выдрав с выдры в тундре гетры, вытру выдрой ядра кедра, вытру гетрой выдре морду ядра в вёдра, выдру в тундру.

За последние несколько лет в сфере архитектуры ПО произошли огромные изменения. Идея единственного монолитного приложения или даже нескольких крупных сервисов, разделяющих общий массив данных, практически стерта из умов и сердец инженеров-практиков во всем мире. Преобладающими инструментами в создании современных бизнес-ориентированных приложений стали автономные микросервисы, событийно-ориентированная архитектура и CQRS. Вдобавок быстрый рост количества подключаемых устройств (мобильных, IoT) многократно увеличивает объем событий, которые система должна оперативно обрабатывать.

В статье рассказываем, за счет чего Apache Kafka работает достаточно быстро для современных проектов.

Давайте начнем с того, что признаем понятие быстрый многогранным, сложным и крайне неоднозначным. На каждую частную интерпретацию этого слова влияют такие показатели как latency, пропускная способность (throughput) и кратковременные задержки (jitter). Слово быстрый также по сути контекстно: индустрия и области применения сами устанавливают нормы и ожидания по производительности.

Apache Kafka заточена под пропускную способность, которая обеспечивается за счет того, что мы жертвуем latency и jitter при сохранении остальных нужных характеристик, таких как надежность, строгий порядок записи и принцип at least once delivery. Когда кто-то при условии, что это хоть сколь-нибудь сведущий человек говорит, что Kafka быстра, можно предположить, что он имеет в виду способность Kafka безопасно накапливать и передавать очень больше количество записей за короткое время.

Если обратиться к истории, Kafka появилась из необходимости LinkedIn эффективно перемещать огромные количества сообщений до нескольких терабайт в час. Latency передачи одного сообщения приписывалось второстепенное значение, также как варьируемости времени задержки. LinkedIn, в конце концов, не финансовый институт, занимающийся высокочастотным трейдингом, и не система промышленного контроля, которая работает в пределах установленных сроков. Kafka может использоваться для внедрения близких к реальному времени систем (near real-time systems), также известных как soft real-time systems.

Пояснение: Для тех, кому не знаком термин реальное время, он означает не быстро, а предсказуемо. В частности, режим реального времени подразумевает наличие жесткой верхней границы, называемой также крайним сроком (deadline), которая определяет время, необходимое для выполнения действия. Если система как целое всякий раз не может соблюсти этих сроков, ее нельзя отнести к системам реального времени. Системы, которые могут работать в пределах вероятного отклонения, называются близкими к реальному времени. Если перевести это на язык пропускной способности, системы реального времени зачастую оказываются медленнее, чем их аналоги, близкие к реальному времени или не относящиеся к реальному времени.

Есть две важные области, из которых Kafka черпает свою скорость, и их надо рассматривать отдельно. Первая связана с низкоуровневой реализацией клиента и брокера. Вторая проистекает из параллелизма потоковой обработки.

Работа брокера

Хранение данных в виде логов

Kafka использует сегментированный журнал, предназначенный только для добавления записей, в значительной степени ограничиваясь последовательным I/O и для чтения, и для операций записи, что работает быстро на самых разнообразных накопителях. Существует широко распространенное заблуждение, что диски работают медленно; однако производительность накопителей информации (особенно жестких дисков) сильно зависит от типов доступа. Производительность случайного I/O на распространенном диске SATA 7200 об/мин на три-четыре порядка ниже по сравнению с последовательным I/O. Более того, современная операционная система предусматривает методы read-ahead (упреждающего чтения) и write-behind (обратной записи), которые предварительно выбирают данные в виде кратных больших блоков и группируют меньшие логические записи в большие физические записи. Из-за этого разница между последовательным I/O и случайным I/O продолжает быть заметной во флэш-памяти и других формах твердотельных энергонезависимых носителей, хотя она гораздо менее значительна по сравнению с жесткими дисками.

Батчинг записей

Последовательный I/O невероятно быстр на большинстве носителей и соизмерим с пиковой производительностью сетевого I/O. На практике это означает, что хорошо спроектированный слой хранения логов на диске будет успевать за сетевым трафиком. Фактически часто узким местом (bottleneck) Kafka становится не диск, а сеть. Поэтому в дополнение к низкоуровневому батчингу, предоставляемому ОС, клиенты и брокеры Kafka собирают в батч многочисленные записи (и чтение, и операции записи) перед отправкой их по сети. Пакетная обработка записей амортизирует издержки на передачу данных в обе стороны, используя более крупные батчи и повышая пропускную способность.

Пакетное сжатие

Влияние пакетирования становится особенно очевидным, когда задействовано сжатие, поскольку сжатие становится более эффективным с увеличением объемов данных. Эффект сжатия может быть достаточно выраженным, особенно если используются текстовые форматы, как JSON, тогда коэффициент сжатия обычно варьируется от 5х до 7х. Более того, пакетная обработка записей по большей части производится как операция на стороне клиента (client-side operation), что передает нагрузку на клиента и оказывает позитивное влияние не только на пропускную способность сети, но и на использование дискового I/O брокера.

Дешевые консюмеры

В отличие от традиционных MQ, которые удаляют сообщения после их прочтения (что приводит к штрафам за случайный I/O), Kafka не удаляет сообщения после прочтения. Вместо этого она независимо отслеживает смещения (offsets) на уровне каждой группы консюмеров (consumer group). Продвижение самих смещений публикуется во внутреннем топике Kafka __consumer_offsets. Это быстро для append-only операции (только добавление). Далее содержание этого топика сжимается в фоновом режиме (с использованием инструмента compaction Kafka), чтобы сохранить последние известные смещения для любой группы консюмеров.

Сравните эту модель с более традиционными брокерами сообщений, которые обычно предлагают несколько противоположных топологий распространения сообщений. С одной стороны, это очередь сообщений надежный транспорт для отправки сообщений point-to-point, без возможности отправки point-to-multipoint. С другой стороны, топик pub-sub позволяет отправку сообщений point-to-multipoint, но делает это за счет надежности. Внедрение надежной модели отправки сообщений point-to-multipoint в традиционной MQ-модели требует поддержания выделенной очереди сообщений для каждого консюмера с отслеживанием состояния. Это создает усиление при read/write amplification. С одной стороны, publisher вынужден писать в много очередей. В качестве альтернативы применяется отправка или fan-out, которая может потреблять записи из одной очереди и записывать их в несколько других, но это только задерживает amplification. С другой стороны, несколько консюмеров создают нагрузку на брокера, представляя собой смесь операций I/O для чтения и записи, как последовательных, так и случайных.

Консюмеры в Kafka дешевы, поскольку они не изменяют файлы журнала (это разрешено только продюсерам или внутренним процессам Kafka). Это означает, что много консюмеров могут одновременно читать из одного и того же топика, не перегружая кластер. Добавление консюмера все еще требует некоторых затрат, но в основном это последовательное чтение с низкой скоростью последовательных операций записи. Поэтому вполне нормально видеть, что один топик используется сразу несколькими консюмерами.

Неизмененные буферизованные операции записи

Еще одна фундаментальная причина производительности Kafka, которую стоит подробнее исследовать: Kafka фактически не вызывает fsync при записи на диск перед подтверждением операции записи. Единственное требование для ACK (подтверждения), чтобы запись была сделана в I/O-буфер. Это малоизвестный, но важный факт. Именно это позволяет Kafka работать так, как если бы это была очередь в памяти. Поскольку по всем параметрам Kafka это очередь в памяти с дисковым бэкапом (disc-backed in-memory queue), ограниченная размером буфера/кэша страниц.

В то же время эта форма операций записи небезопасна, поскольку сбой реплики может привести к потере данных, даже несмотря на то что запись, казалось бы, была подтверждена. Другими словами, в отличие от, скажем, реляционной базы данных, подтверждение записи само по себе не означает надежности. Надежность Kafka обеспечивается запуском нескольких синхронизированных реплик. Даже если одна даст сбой, другие (при условии, что их больше одной) продолжат работать при условии, что сбой не коррелирован (то есть несколько реплик не дают сбой одновременно из-за общего сбоя на директивном уровне). Таким образом, сочетание неблокирующего подхода к I/O без использования fsync, плюс дублирующие синхронизируемые реплики приводит Kafka к сочетанию высокой пропускной способности, надежности и доступности.

Оптимизация на стороне клиента

Большинство баз данных, очередей и других хранилищ созданы вокруг понятия всемогущего сервера (или кластера серверов) и довольно небольших клиентов, которые взаимодействуют с сервером(ами) по хорошо известному протоколу. Распространено мнение, что клиентские реализации гораздо проще серверных. В результате сервер берет на себя основную нагрузку, а клиенты выступают лишь как интерфейсы между кодом приложения и сервером.

Kafka использует другой подход к дизайну клиента. Прежде чем записи попадут на сервер, значительное количество работы производится на клиенте. Сюда относится аккумулирование записей, хеширование ключей записи, чтобы они оказались в правильной партиции, проверка checksum записей и сжатие батча записей. Клиент знает о метаданных кластера и периодически обновляет эти метаданные, чтобы не отставать от изменений в топологии брокера. Это позволяет клиенту принимать решения о пересылке на нижнем уровне. Вместо того чтобы слепо пересылать запись в кластер и полагаться на то, что последний перешлет их на соответствующий брокерский узел, клиент пересылает записи непосредственно мастерам партиций. Похожим образом клиенты консюмеры могут принимать разумные решения, когда ищут источники записей, потенциально используя реплики, которые географически ближе к клиенту при выдаче запросов на чтение. (Эта функция была добавлена в Kafka позднее и доступна с версии 2.4.0.)

Zero-copy

Один из обычных источников неэффективности копирование байтовых данных между буферами. Kafka использует бинарный формат сообщений, который является общим для продюсера, брокера и консюмера, поэтому фрагменты данных могут передаваться напрямую без изменений, даже если они подверглись сжатию. Хотя устранение структурных различий между общающимися сторонами является важным шагом, само по себе оно не избегает копирования данных.

Kafka решает эту проблему на Linux и UNIX при помощи Javas NIO, в частности метод transferTo() класса java.nio.channels.FileChannel. Этот метод позволяет передавать байты из исходного канала в канал-приемник без привлечения приложения как посредника передачи. Чтобы оценить перемены, привнесенные NIO, рассмотрим традиционный подход, в котором исходный канал считывается в байтовый буфер и затем записывается в канал-приемник как две отдельные операции:

File.read(fileDesc, buf, len);Socket.send(socket, buf, len);

Графически это можно представить следующим образом:

Хотя это выглядит достаточно просто, внутренне операция копирования требует четырех переключений контекстов между пользовательским режимом и режимом ядра, и данные копируются четыре раза до завершения операции. Представленная ниже диаграмма показывает контекстные переключения на каждом этапе.

Если посмотреть на это более подробно:

1. Начальный read() вызывает переключение контекста из пользовательского режима в режим ядра. Файл прочитан, а его содержимое копируется в буфер в адресном пространстве ядра при помощи механизма DMA (Direct Memory Access, прямой доступ к памяти). Это не тот же самый буфер, который использовался во фрагменте кода.

   
   

2. Перед возвратом из read() буфер ядра копируется в буфер пользовательского пространства. Здесь наше приложение может прочитать содержимое файла.

   
   

3. Последующий send()переключится обратно в режим ядра, скопировав буфер пользовательского пространства в адресное пространство ядра на этот раз в другой буфер, связанный с сокетом назначения. За кулисами за дело берется механизм DMA, асинхронно копируя данные из буфера ядра в стек протокола. Метод send() ждет этого после возврата.

   
   

4. Метод/вызов send() возвращается, переключаясь обратно в контекст пользовательского пространства.

   
   

Несмотря на недостатки переключения контекстов и дополнительное копирование, промежуточный буфер ядра во многих случаях может улучшить производительность. Он может выступать в качестве кеша упреждающего чтения, асинхронно выбирая блоки, тем самым с опережением выполняя запросы приложений. Однако когда количество запрашиваемых данных значительно превышает объем буфера ядра, буфер ядра становится узким местом производительности системы. Вместо того чтобы напрямую копировать данные, он вынуждает систему переключаться между режимами пользователя и ядра на всем процессе передачи данных.

Для сравнения, подход с zero-copy обрабатывается за одну операцию. Фрагмент из предыдущего примера можно переписать в одну строчку:

fileDesc.transferTo(offset, len, socket);

Подход с zero-copy проиллюстрирован на схеме ниже:

В этой модели количество переключений контекста снижено до одного. Если конкретно, то метод transferTo() указывает блочному устройству считывать данные в буфер чтения при помощи механизма DMA. Этот буфер затем копируется в другой буфер ядра для подготовки к записи в сокет. Наконец, буфер сокета копируется в буфер NIC с помощью DMA.

В результате мы снизили число копирований с четырех до трех, и только одно копирование использует CPU. Мы также сократили количество переключений контекста с четырех до двух.

Это большое улучшение, но это еще не zero-copy. Последнее может быть достигнуто в виде дальнейшей оптимизации при работе с ядром Linux версии 2.4 и выше, а также на сетевых картах, которые поддерживают операцию gather. Иллюстрация ниже.

Вызов метода transferTo() заставляет устройство считывать данные в буфер ядра при помощи механизма DMA, как в предыдущем примере. Однако при операции gather отсутствует копирование между буфером чтения и буфером сокета. Вместо этого NIC получает указатель на буфер чтения вместе со смещением и длиной. Буфер чтения очищается DMA. Ни в один момент CPU не вовлечен в копирование буферов.

Сравнение традиционного подхода и подхода с zero-copy на файлах различного размера от нескольких мегабайт до гигабайта показывают двух- или трехкратный рост производительности в пользу zero-copy. Но еще более впечатляет то, что Kafka добивается этого, используя простой JMV без собственных библиотек или JNI кода.

Избегание сборки мусора

Активное использование каналов, собственных буферов и кэша страниц рождает еще одно дополнительное преимущество снижение нагрузки на сбор мусора (GC). Например, запуск Kafka на машине с 32 Гб RAM приведет к тому, что 28-32 Гб будут использоваться как кэш страниц, полностью выходя за пределы диапазона действия GC. Разница в пропускной способности минимальна (в районе нескольких десятых процента), поскольку пропускная способность правильно настроенного GC может быть достаточно высокой, особенно когда он работает с краткосрочными объектами. Реальные успехи лежат в сфере снижения джиттера. Избегая GC, брокеры с меньшей вероятностью испытают паузу, которая сможет повлиять на клиента, увеличив задержку сквозной доставки записей.

Справедливости ради стоит заметить, что избегание GC стало сейчас меньшей проблемой по сравнению с тем, когда Kafka только зарождалась. Современные GC, как Shenandoah и ZGC масштабируются до огромных многотерабайтных куч и имеют настраиваемую по длительности паузу для наихудшего случая вплоть до однозначных миллисекунд. В наши дни можно часто увидеть, как приложения на основе JVM, которые используют большие кэши на основе кучи, превосходят проекты без куч.

Потоковый параллелизм

Эффективность журнально-структурированного I/O один из важнейших аспектов производительности, в основном влияющий на операции записи. Для производительности чтения фундаментальными является то, как Kafka трактует параллелизм в структуре темы и экосистемы консюмеров. Их комбинация создает очень высокую общую пропускную способность для сквозного обмена сообщениями. Параллельное выполнение встроено в ее схему разделения и работу групп консюмеров, что является тем самым механизмом балансировки нагрузки внутри Kafka, распределяя задания разделения примерно равным образом между отдельными экземплярами консюмеров в группе. Сравните это с более традиционным MQ: в эквивалентной настройке RabbitMQ множественные одновременные консюмеры могут читать из очереди, построенной по циклическому алгоритму, но при этом они теряют понятие упорядочивания сообщений.

Механизм разделения также позволяет горизонтально масштабировать брокеров Kafka. У каждого раздела есть отдельный лидер; поэтому каждый нетривиальный топик (со множеством разделов) может использовать весь кластер брокера для операций записи. И это еще одно различие между Kafka и очередью сообщений. Тогда как последняя использует кластеризацию для обеспечения доступности, Kafka действительно распределяет нагрузку между брокерами ради поддержания доступности, надежности и пропускной способности.

При публикации записи продюсер определяет раздел, мы предполагаем сейчас, что вы публикуете в топик с несколькими разделами. (Если у кого-то топик с одним разделом, то такой проблемы нет). Это можно сделать либо напрямую, указав индекс раздела, либо опосредованно с помощью ключа записи, который детерминированно хеширует в постоянный (то есть один и тот же каждый раз) индекс раздела. Записи, имеющие один и тот же хеш, гарантированно займут один и тот же раздел. Предположим, что в теме несколько разделов, тогда записи с другим ключом, скорее всего, окажутся в других разделах. Однако из-за коллизий хеш-функций записи с разными хешами могут также оказаться в одном разделе. Такова природа хеширования. Если вы понимаете, как работает хеш-таблица, то здесь все то же самое.

Фактическая обработка записей производится консюмерами, работающими в рамках (опциональной) группы консюмеров. Kafka гарантирует, что раздел может быть закреплен максимально за одним консюмером в его группе консюмеров. (Мы говорим максимум, предусматривая случай, когда все консюмеры находятся оффлайн). Когда первый консюмер в группе подписывается на топик, он получает все разделы по этом топике. Когда позднее на этот топик подписывается второй консюмер, он получит примерно половину разделов, освобождая первого консюмера от половины его предыдущей нагрузки. Это позволяет обрабатывать поток событий параллельно, при необходимости добавляя консюмеров (в идеале используя механизм автоматического масштабирования), при условии что вы правильно разбили поток событий.

Контроль пропускной способности записи осуществляется двумя способами:

1. Схема разбиения топика. Топики должны быть разделены, чтобы максимизировать количество независимых суб-потоков событий. Другими словами, порядок записи следует сохранять только там, где это абсолютно необходимо. Если любые две записи не связаны обоснованно в причинно-следственном смысле, они не должны быть привязаны к одному и тому же разделу. Это подразумевает использование разных ключей, так как Kafka использует ключ записи в качестве источника хеширования, чтобы получить согласованную карту разделов.

   
   

2. Количество консюмеров в группе. Вы можете увеличить количество консюмеров так, чтобы оно соответствовало нагрузке от входящих сообщений, вплоть до количества разделов в теме. (Можно иметь и больше консюмеров, если хотите, но счетчик разделов в таком случае ограничит количество активных консюмеров, которые получат минимум один назначенный раздел; оставшиеся консюмеры будут находиться в режиме ожидания). Обратите внимание, что консюмером может быть процесс или поток. В зависимости от типа рабочей нагрузки, которую выполняет консюмер, вы можете использовать несколько потоков из отдельных консюмеров или записи процессов в пуле потоков.

   
   

Если вы задавались вопросом, быстра ли Kafka, как она достигает своих знаменитых характеристик производительности, может ли она масштабироваться к вашим практическим кейсам, вы должны были, надеюсь, уже получить все ответы, которые вам были нужны.

Чтобы окончательно все прояснить, Kafka не самое быстрое (если смотреть на скорость с точки зрения пропускной способности) межплатформенное ПО для обмена сообщениями: есть другие платформы, с бОльшими пропускными способностями, какие-то программные, другие реализованы на железе. И также она не предлагает лучший компромисс между пропускной способностью и latency. Например, Apache Pulsar многообещающая технология, которая может масштабироваться и достигает лучшего профиля пропускной способности и latency, в то же время предлагая идентичный Kafka порядок и гарантии надежности. Основная причина для выбора именно Kafka это то, что как целостная экосистема она в общем-то по прежнему не знает себе равных. Она показывает отличную производительность, в то же время предлагая среду, которая одновременно устойчивая, сложившаяся и углубляющаяся: несмотря на свои размеры Kafka продолжает расти завидными темпами.

Разработчики и специалисты по сопровождению Kafka проделали потрясающую работу по разработке ориентированного на производительность решения. Лишь малая часть ее конструктивных элементов ощущаются придуманными задним числом или ситуативными. Начиная от переноса работы на клиентов и до лог-структурированного хранилища на брокере, пакетирования, сжатия, zero-copy I/O и потокового параллелизма Kafka бросает вызов практически любому ориентированному на сообщения межплатформенному ПО, коммерческому или с открытым исходным кодом. Еще больше впечатляет то, что она делает это без ущерба для надежности, порядка записи и модели at least once delivery.

Kafka не самая простая платформа для обмена сообщениями, в ней есть, что поизучать. Прежде чем уверенно разрабатывать и строить высокопроизводительные, управляемые событиями системы, стоит разобраться с понятиями полного и частичного порядка, тем, разделов, консюмеров и групп консюмеров. Объем материала очень велик, но результат, безусловно, стоит затраченных вами усилий.

От редакции: Если вам интересно поизучать Kafka, Слёрм готовит новый видеокурс от инженеров компаний Авито и Stripe. Курс сейчас находится в разработке. Можно оставить заявку, чтобы купить по цене предзаказа и повлиять на программу.

Apache Kafka стала ведущей платформой для асинхронной коммуникации между микросервисами. В ней есть мощные функции, которые позволяют строить устойчивые к ошибкам и отказам асинхронные архитектуры.

В то же время нужно предвидеть потенциальные ловушки. Неспособность заранее распознать проблемы, которые могут (нет, будут) возникать, приведет к тому, что у нас будут уязвимые к ошибкам и искажению данных системы.

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

Краткий обзор Kafka

Любой читатель этой статьи, вероятно, хоть немного знаком с Kafka. Про то, что такое Kafka и как с ней работать, есть разные подробные материалы. С учетом сказанного мы кратко остановимся на важных для нашего обсуждения понятиях.

Журнал событий, продюсеры и консюмеры

Kafka система для обработки потоков информации. Концептуально мы понимаем Kafka как общность трех базовых компонентов:

Журнал событий (event log), в который публикуются сообщения.
Продюсеры (producers, publishers), публикуют сообщения в журнал событий.
Консюмеры (consumers), потребляют сообщения из журнала событий.

В отличие от традиционных очередей сообщений, как в RabbitMQ, в Kafka именно консюмеры определяют, когда прочитать сообщение (т.е. Kafka использует скорее pull, нежели push модель). У каждого сообщения есть смещение (offset), и каждый консюмер отслеживает (или фиксирует) смещение последнего потребленного сообщения. Это позволяет консюмерам затем запрашивать следующее сообщение через смещение этого сообщения.

Топики

Журнал событий разделен на топики (topics), каждый из которых определяет один тип сообщений, которые в него публикуются. И хотя от нас, инженеров, зависит определение топиков, нам стоит помнить некоторые простые и надежные правила:

• Каждый топик описывает событие, о котором должны знать другие сервисы.
• Каждый топик должен определять единую схему, которой должно соответствовать каждое сообщение.

Партиции и ключи партиций

Темы далее разделяются на партиции (partitions). Партиции дают возможность потреблять сообщения параллельно. Kafka позволяет детерминированно назначать сообщения в партиции при помощи ключа партиции (partition key). Ключ партиции это часть данных (обычно некоторый атрибут самого сообщения, например, идентификатор), к которой применяется алгоритм для определения партиции.

Здесь мы назначаем ключом партиции поле UUID сообщений. Продюсер применяет алгоритм (например, модифицирует каждое значение UUID по количеству партиций), чтобы отнести каждое сообщение к какой-то партиции.

Использование ключей партиций таким образом позволяет нам удостовериться в том, что каждое сообщение, сопоставляемое с определенным идентификатором, публикуется в одной партиции.

Также стоит отметить, что несколько консюмеров могут быть объединены в одну группу консьюмеров. Kafka обеспечит прочтение любого сообщения в данной партиции одним и тем же количеством консюмеров в рамках группы.

Использование Kafka с микросервисами

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

Передача сообщения через ограниченный контекст

Когда мы только начали создавать микросервисы, многие из нас сразу приняли централизованную модель. У каждого фрагмента данных был единственный микросервис (т.е. единственный источник истины), в котором этот фрагмент находился. Если какому-то другому микросервису был нужен доступ к этим данным, он делал синхронный вызов для их получения.

Этот подход привел к ряду проблем, включая длинные синхронные цепочки вызовов, единые точки отказа, снижение автономии команды и т.д.

В итоге мы узнали лучший способ. В современных зрелых архитектурах мы разбиваем коммуникацию на обработку команд и обработку событий.

Обработка команд обычно выполняется в одном ограниченном контексте (bounded context) и по-прежнему обычно включает синхронную коммуникацию.

События же обычно генерируются службами в одном ограниченном контексте и асинхронно публикуются в Kafka для потребления службами в других ограниченных контекстах.

На схеме слева представлено то, как мы раньше проектировали связь между микросервисами: сервис в рамках одного ограниченного контекста (представлен прямоугольником с пунктирной линией) получает синхронные вызовы от сервисов в рамках других ограниченных контекстов. Справа представлено то, как мы делаем это сейчас: сервис в рамках одного ограниченного контекста публикует события, сервисы из других ограниченных контекстов потребляют эти события в своем темпе.

Назовем этот вариант трансграничной публикацией событий (cross-boundary event publishing).

Производя трансграничную публикацию событий, мы должны публиковать агрегаты (aggregates). Агрегаты это независимые группы сущностей, каждая из которых рассматриваются как единичные атомарные сущности. У каждого агрегата есть корневой объект, а также некоторое количество подчиненных объектов, которые предоставляют дополнительные данные.

В нашем примере мы можем определить агрегат UserAccount. Его корневым объектом будет, вероятно, сущность User (содержащая в глобальном масштабе уникальный ID: имя, фамилия, дата рождения и т.д.). Там также могут содержаться сущности, представляющие фрагменты контактной информации (EmailAddress, PhoneNumber и т.д.). Что важно, агрегат примет уникальный в глобальном масштабе ID как свой собственный ID.

Когда наш UserAccount публикует сообщение, информационное наполнение этого сообщения будет некоей репрезентацией (JSON или Avro, например) агрегата UserAccount. Важно, что сервис определит уникальный ID учетной записи пользователя как ключ партиции. Это обеспечит в будущем публикацию изменений в любых сущностях любого UserAccount в одну и ту же партицию.

Что происходит, когда возникает проблема?

Хотя трансграничная публикация событий с помощью Kafka оказывается достаточно элегантной, это распределенная система. Поэтому многое может пойти не так. Мы рассмотрим, наверное, самую типичную, мучительную проблему: консюмер не может обработать потребленное сообщение.

И что теперь делать?

Идентифицировать это как проблему

Первое, что команды делают не так, это просто не распознают это как потенциальную проблему. Время от времени будет случаться сбой сообщений, и нам придется разработать стратегию решения этой проблемы до того, как она по-настоящему случится.

Поэтому первым шагом будет понять, что эта проблема произойдет и разработать решение для ее устранения. Если мы дошли до этого пункта, нам нужно себя поздравить. Теперь встает большой вопрос: что с этим делать?

Разве нельзя просто повторно это сообщение передать (retry)?

По умолчанию, если консюмер не производит успешного потребления сообщения (т.е. консюмер не может произвести текущее смещение), он будет снова и снова возвращаться к этому сообщению. Почему же мы не можем позволить этой модели по умолчанию работать и возвращаться к сообщению, пока не будет успеха?

Проблема в том, что успешной обработки может не произойти никогда. По крайней мере без какого бы то ни было ручного вмешательства. Консюмер поэтому никогда не перейдет к последующим сообщениям, и обработка сообщений зайдет в тупик.

Хорошо, а нельзя ли просто пропустить сообщение?

Мы обычно позволяем синхронным запросам не заканчиваться успехом. Например, ПУБЛИКАЦИЯ создать пользователя, созданная в нашем UserAccount, может содержать неверные или неполные данные. В этом случае мы просто можем выдать код ошибки (например, HTTP 400) и попросить вызывающего оператора попытаться еще раз.

И хотя это не идеальный выход, он не приведет ни к каким продолжительным проблемам в целостности наших данных. Эта ПУБЛИКАЦИЯ представляет команду, что-то, что еще не произошло. Даже если мы позволим этой публикации завершиться без успеха, наши данные останутся в согласованном состоянии.

Это не тот случай, когда мы отбрасываем сообщения. Сообщения представляют собой события, которые уже произошли. Любой консюмер, игнорирующий события, будет перманентно не согласован с вышестоящим сервисом, который сформулировал событие.

И все это ради того, чтобы сказать, мы не хотим пропускать сообщения.

И как же нам решить проблему?

Это не самая легкая для нас проблема. Поэтому когда мы осознали необходимость ее решить, нам, возможно, придется посмотреть в интернете, как же ее решить. Что приводит ко второй проблеме: там можно найти советы, которым лучше не следовать.

Retry topic: популярное ходовое решение

Одно из популярных решений, которое вы найдете на просторах интернета, включает в себя топик повтора (retry topic). Мелкие детали разнятся от реализации к реализации, но общая канва такова:

1. Консюмер делает попытку потребить сообщение из главного топика.
2. Если первая попытка не удалась, консюмер публикует сообщение в топик повтора и затем фиксирует смещение сообщения, чтобы иметь возможность продолжить работу со следующим сообщением.
3. На топик повтора подписан консюмер повтора, у которого та же самая логика, что и у основного консюмера. Этот консюмер вводит короткую отсрочку между попытками потребить сообщение. И если этот консюмер также не может удачно потребить сообщение, он публикует сообщение во второй топик повтора и фиксирует смещение сообщения.
4. Так это продолжается с еще некоторым количеством повторных топиков и консюмеров, в каждом случае с увеличивающейся отсрочкой (что служит стратегией отката (backoff strategy)). В конце концов, когда сообщение не удается обработать последнему консюмеру повтора, оно отправляется в очередь отвергаемых сообщений (dead letter queue, DLQ), где его вручную сортируют инженеры.

Концептуально шаблон retry-topic (топик повтора) определяет количество топиков, на которые переведут необработанные сообщения. Если консюмер главного топика потребляет сообщение, которое он не может обработать, он опубликует это сообщение в топике повтора 1 и совершает текущее смещение, чтобы освободиться для следующих сообщений. Консюмер из топика повтора будет клоном основного консюмера, с тем лишь отличием, что он в случае неудачи будет публиковать сообщение в новый топик повтора. В итоге, если последний консюмер не смог обработать сообщение, он опубликует сообщение в очередь отвергаемых сообщений (dead letter queue, DLQ)

В чем же проблема?

На первый взгляд, этот подход кажется обоснованным. И в реальности он нормально сработает во многих случаях. Проблема в том, что это решение не подойдет в качестве универсального. На самом деле есть некоторые случаи, как наша трансграничная публикация событий, для которых этот подход будет опасным.

Он не берет в расчет различия типов проблем

Первая проблема: он не берет в расчет, что в большинстве случаев будет встречаться одна из двух причин неудачного потребления событий: исправимая ошибка и неисправимая ошибка.

Исправимые ошибки (recoverable errors) ошибки, которые в случае, если мы достаточно количество раз предпримем повтор, в итоге будут исправлены. Простым примером может стать консюмер, сохраняющий данные в базу. Если база данных будет временно недоступна, консюмер даст сбой, когда придет следующее сообщение. Как только база данных снова станет доступна, тогда консюмер снова сможет обрабатывать сообщение.

По-другому на это можно посмотреть следующим образом: исправима та ошибка, корневая причина которой является внешней по отношению к сообщению и консюмеру. Как только решается эта проблема, наш консюмер продолжит работу, как будто ничего не произошло. (И вот здесь как раз кроется частый источник путаницы. Термин исправимый не означает, что само приложение в нашем случае консюмер производит исправление. Как раз наоборот, именно какой-то внешний ресурс в нашем случае база данных дает сбой и в итоге исправляется.)

Штука в том, что исправимые ошибки поразят все без исключения сообщения в теме. Вспомним, что все сообщения в теме должны соответствовать одной схеме и представлять один тип данных. Аналогично и наш консюмер будет проводить то же самое действие(я) для каждого события в теме. Поэтому если сообщение А должно дать сбой из-за отказа базы данных, тогда то же самое сделают сообщение В, сообщение С и т.д.

Неисправимые ошибки(non-recoverable errors) ошибки, которые все равно будут случаться, независимо от того, сколько раз мы попробуем повторно выполнить операцию. Например, пропуск поля в сообщении может привести к NullPointerException. Или из-за поля со специальным символом сообщение будет отображено как не поддающееся синтаксическому анализу.

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

Также, в отличие от исправимых ошибок, чтобы решить неисправимые ошибки, надо исправить сам консюмер (никогда не исправляйте сами сообщения: это неизменяемые записи!) Например, мы можем исправить наш консюмер, чтобы все значения NULLS обрабатывались должным образом, а затем снова его развернуть.

И какое все это имеет отношение к решению с топиками повторов?

Для начала, это не будет особо полезным при исправимых ошибках. Мы помним, что до разрешения внешней проблемы исправимая ошибка будет затрагивать каждое сообщение, не только текущее. Так что, конечно, направление сообщения со сбоем в топик повтора освободит путь для следующего сообщения. Но это сообщение также даст сбой, и следующее после него, и после него Мы просто можем предоставить консюмеру возможность самому повторять операцию, пока проблема не разрешится.

А что насчет неисправимых ошибок? Очереди повтора могут помочь в этом случае. Если одно проблемное сообщение задерживает потребление всех последующих сообщений, то уборка его с пути определенно разблокирует нашего консюмера (конечно, наличие нескольких топиков повтора становится излишним).

Однако хотя очередь повторов может помочь разблокировать консюмеров, устранив сообщения с неисправимыми ошибками, она может привести к еще более серьезным проблемам. Давайте читать дальше, чтобы выяснить почему

Он игнорирует очередность

Кратко повторим некоторые важные аспекты трансграничной публикации событий. После обработки команды в нашем ограниченном контексте мы публикуем соответствующее событие в топик Kafka. Что важно, мы указываем ID агрегата в качестве ключа партиции.

Почему это важно? Это гарантирует, что изменения в любом отдельно взятом агрегате всегда публикуются в одну и ту же партицию.

Хорошо. Почему же тогда это важно? Публикация событий в одну и ту же партицию гарантирует их обработку в той последовательности, в которой они произошли. Если изменения в одном и том же агрегате произведены с небольшим интервалом, а возникающие в связи с ними события записаны в разные партиции, может возникнуть подобие гонки, в ходе которой консюмер потребляет второе изменение прежде первого. А это приводит к противоречивости данных.

Давайте рассмотрим небольшой пример. Наш ограниченный контекст User предлагает приложение, позволяющее пользователям менять свои имена. Пользователь А меняет свое имя с Zoey на Zoё и затем сразу же меняет его на Zoiee. Если бы нам не была важна очередность, низовой консюмер (скажем, в ограниченном контексте Login) первым может обработать изменение на Zoiee, а сразу после этого переписать его на Zoё.

Теперь данные Login рассинхронизированы с данными User. Более того, каждый раз при входе на наш сайт Zoiee будет видеть приветствие: Добро пожаловать, Zoё!

И это настоящая проблема с retry topic. Они оставляют наши консюмеры уязвимыми к обработке событий не в порядке очередности. Если на консьюмера повлияет временный сбой базы данных, пока он обрабатывает изменение Zoё, он переправит сообщение в топик повтора, чтобы обработать его позднее. Если сбой в базе данных исправлен к тому времени, как появляется изменение Zoiee, тогда это сообщение будет успешно обработано первым, а позднее перезаписано изменением Zoё.

Для понятности пример с Zoiee/Zoё так прост. В реальности обработка событий в порядке очередности может привести к целому ряду проблем с искажением данных. Что хуже, эти проблемы сперва остаются незамеченными. Они обычно приводят к искажению данных, которое остается какое-то время никем не замеченным, но постепенно увеличивается. Часто к тому времени, когда мы понимаем, что произошло, большие фрагменты информации уже поражены.

В каких же случаях retry topic может быть полезен?

Стоит прояснить, что это не всегда плохой вариант. Несомненно, есть случаи, когда топики повторов оптимально подходят. Если конкретно, этот шаблон применяем, когда задача консюмера собирать неизменяемые записи. Такие примеры могут включать:

• Консюмера, который обрабатывает потоки активности на веб-сайте для создания отчетов.
• Консюмера, добавляющего транзакции в журнал (при условии, что эти транзакции не нужно отслеживать в определенном порядке.
• Консюмера, который извлекает данные из других источников данных.

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

Все же небольшое предостережение

Даже в случае таких вариантов использования нам все равно следует действовать с осторожностью. Создание такого решения сложно и затратно. Поэтому как организация мы не хотим писать новое решение для каждого нового консюмера. Вместо этого мы захотим создать одно решение библиотеку, контейнер и т.д. которое можно использовать в разных сервисах.

И здесь кроется дополнительный повод для беспокойства. Мы можем создать решение с топиком повторов для релевантных консюмеров. К сожалению, через некоторое время это решение проникнет в консюмеры трансграничной публикации событий. Команды, которые занимаются теми консюмерами, могут не знать о рисках. И как мы обсуждали ранее, они просто не заметят проявления проблем, пока не произойдет существенного искажения данных.

Поэтому прежде чем мы внедрим решение с топиком повторов, мы должны быть на 100% уверены, что либо:

• Наш бизнес таков, что у нас никогда не будет консюмеров, которые обновляют существующие данные, или
• У нас есть незыблемые средства управления, чтобы не допустить внедрения решения с топиком повторов для таких консюмеров.

Как мы можем исправить шаблон?

Учитывая, что шаблон с топиком повтора не может быть приемлемым решением для трансграничной публикации событий, можем ли мы его адаптировать так, чтобы это стало возможным?

Изначально я планировал, чтобы эта статья содержала решение, но в процессе я осознал, что нет единого решения на все случаи. Поэтому вместо первоначального плана мы обсудим некоторые вещи, которые стоит учитывать.

Снятие неоднозначности типов ошибок

Наша жизнь будет гораздо проще, если мы сможем снимать неоднозначность между исправимыми и неисправимыми ошибками. Например, если наш консюмер начинает испытывать исправимые ошибки, то топики повторов становятся излишними.

Поэтому мы можем попытаться определить, с каким типом ошибки мы имеем дело:

void processMessage(KafkaMessage km) { try {   Message m = km.getMessage();   transformAndSave(m); } catch (Throwable t) {   if (isRecoverable(t)) {     // ...   } else {     // ...   } }}

В приведенном выше примере псевдо-кода на Java isRecoverable() будет использовать подход белого списка (whitelist approach) для определения, представляет ли t исправимую ошибку. Другими словами, он проверяет t, чтобы определить соответствует ли она любой другой известной исправимой ошибке (скажем, ошибке соединения SQL или таймаут клиента REsT), и возвращает true, если это так. Если нет, возвращает false. Это поможет предотвратить нескончаемую блокировку неисправимых ошибок.

Признаемся, что снимать неоднозначность между исправимыми и неисправимыми ошибками может быть непросто. Например, SQLException может показывать отказ базы данных (исправимый) или нарушение ограничения (неисправимое). При возникновении сомнений нам, наверное, стоит предположить, что ошибка неисправима, что повышает риск отправить хорошие сообщения в stash topic (назовем его топик-загашник), что отложит их обработку Но в то же время мы избежим нечаянного забуксовывания, когда консюмер будет бесконечно пытаться обработать неисправимую ошибку.

Повторная обработка исправимых ошибок консюмером

Как мы уже говорили, в случае с исправимыми ошибками мало смысла отправлять сообщение в топик повторов. Мы просто освободим очередь, чтобы дать место для провала следующему сообщению. Вместо этого консюмер просто может пытаться заново, пока состояние не исправится.

Конечно, наличие исправимой ошибки свидетельствует о наличии проблемы с внешним ресурсом.

Забивание этого ресурса запросами не улучшит ситуацию. Поэтому мы захотим применить стратегию отката для наших повторных попыток. Псевдо-Java код теперь выглядел бы как-то так:

void processMessage(KafkaMessage km) { try {   Message m = km.getMessage();   transformAndSave(m); } catch (Throwable t) {   if (isRecoverable(t)) {     doWithRetry(m, Backoff.EXPONENTIAL, this::transformAndSave);   } else {     // ...   } }}

(Заметьте, любой механизм отката должен быть настроен так, чтобы предупреждать нас, когда он достигает определенного предела, уведомляя нас о потенциально серьезной проблеме).

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

С другой стороны, когда наш консюмер встречается с неисправимой ошибкой, нам, вероятно, хочется убрать это сообщение подальше, чтобы освободить последующие сообщения. Но поможет ли в этом случае множество топиков повторов? Не совсем. Просто наше сообщение не сможет быть обработано n раз, прежде чем отправиться в DLQ. Тогда почему бы сразу его туда не отправить?

Также как и с топиком повтора, у этого топика-загашника будет свой консюмер, идентичный основному консюмеру. Но так же как DLQ, этот консюмер не всегда будет потреблять сообщения. Он будет это делать только в случае явной в этом необходимости.

Подумаем об очередности

Давайте посмотрим, где мы с вами находимся в плане очередности. Используем снова пример User/Login. Консюмер Login может получить ошибку, пытаясь обработать символ ё в имени Zoё. Распознав его как неисправимую ошибку, консюмер откладывает это сообщение в сторону и продолжает со следующими. Вскоре он дойдет до сообщения Zoiee и удачно его обработает.

Сообщение Zoё было отложено в загашник, а сообщение Zoiee удачно обработано. На некоторое время данные между двумя ограниченными контекстами единообразны.

Чуть позже наша команда исправляет консюмера, чтобы он мог правильно обрабатывать специальные символы, и заново его развертывает. После этого мы заново публикуем сообщение Zoё в консюмера, и теперь он безошибочно его обрабатывает.

Когда обновленный консюмер позднее обрабатывает отложенное сообщение Zoё, данные между двумя ограниченными контекстами не согласуются. Поэтому пока ограниченный контекст User думает о пользователе как о Zoiee, ограниченный контекст Login знает ее как Zoё.

Очевидно, у нас не получилось соблюсти очередность. Zoё была обработана консюмером Login после Zoiee, хотя должно было быть наоборот. Мы можем начать откладывать в загашник каждое сообщение после откладывания одного, но в таком случае мы быстро окажемся в тупике.

Однако если наш консюмер может отслеживать ID сообщений, которые он отложил, он может также откладывать все сообщения с тем же самым ID. Другими словами, нам не надо соблюдать очередность всех сообщений А только сообщений с единственным агрегатом.

Как только мы получили уведомление об отложенных в загашнике сообщениях, мы можем отменить развертывание нашего консюмера и исправить его код (важно: никогда не изменяйте само сообщение, сообщения представляют собой неизменяемые события!). После исправления и тестирования консюмера мы можем заново ввести его в эксплуатацию. Конечно, перед возобновлением потребления основного топика нам надо будет с особым внимание отнестись к обработке всех отложенных сообщений из топика-загашника. Таким образом мы продолжим поддерживать очередность. По этой причине мы сначала запустим консюмера загашника, и только после того как он завершит (имеем в виду все единицы в группе консюмеров, если мы используем несколько консюмеров), мы отменим его развертывание и развернем основного консюмера.

Мы также должны понимать, что пока наш исправленный консюмер обрабатывает отложенные сообщения, он может столкнуться с дополнительными ошибками. В этом случае его стратегия по решению ошибок будет такой, как мы уже обсуждали:

• если ошибка исправимая, он повторит с откатами;
• если ошибка неисправимая, он отложит данное сообщение и продолжит со следующим.

Для этих целей, возможно, нам стоит подумать о создании второго топика-загашника.

Согласны на некоторую несогласованность данных?

Создание такой системы может быть довольно сложной задачей. Ее сложно строить, тестировать и обслуживать. Поэтому некоторые организации могут решить определить вероятность появления несогласованных данных и решить, могут ли они жить с таким риском.

Во многих случая такие организации могут задействовать механизм сверки данных, чтобы их данные в итоге (и это достаточно длительное в итоге) становились согласованными. Для этого существует много стратегий, но все они лежат за пределами охвата этой статьи.

Заключение

Если вам кажется, что работа с повторными попытками обработки сложна, вам не кажется. Это так и есть. Особенно если сравнивать ее с относительной элегантностью Kafka, когда все идет правильно. Любое решение, которое мы построим будь то топики повторов, топики отложенных сообщений или любые другие решения будут гораздо более сложными, чем нам бы того хотелось.

К сожалению, это не то, что мы можем игнорировать, если мы надеемся построить устойчивые асинхронные коммуникационные потоки между нашими микросервисами.

В этой статье представлено популярное решение со всеми недостатками и пищей для размышления, когда вы создаете альтернативное решение. Главное, чтобы построить правильное решение, нужно держать в голове пару вещей:

• Понимать, что представляет из себя Kafka в вопросе топиков, партиций и ключей партиций.
• Учитывать отличия между исправимыми и неисправимыми ошибками.
• Использовать такие паттерны проекта, как ограниченные контексты и агрегаты.
• Четко представлять случаи использования Kafka в нашей организации как в настоящем, так и в перспективе. Мы только перемещаем независимые записи? В этом случае нам особо не надо заботиться об очередности. Или мы распространяем события, которые представляют собой изменения в наших данных? В этом случае очередность совершенно необходима.
• Обдумать, готовы ли мы жить с несогласованностью какого бы то ни было объема данных.

От редакции:
Для тех, кому интересно изучение Apache Kafka, Слёрм готовит видеокурс от инженеров компаний Авито и Stripe. Курс сейчас находится в разработке. Можно оставить заявку, чтобы купить по цене предзаказа и повлиять на программу.*

Источники:

• https://dzone.com/articles/creating-apache-kafka-topics-dynamically-as-part-o
• https://quarkus.io/blog/kafka-failure-strategy/
• https://eng.uber.com/reliable-reprocessing/
• https://jaceklaskowski.gitbooks.io/apache-kafka/content/kafka-topics.html
• https://www.red-gate.com/simple-talk/sql/bi/reconciling-data-across-systems-using-reconciliation-hub/

Открываем предзаказ продвинутого курса по Apache Kafka.

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

Курс подойдет опытным инженерам.

Программа

Базовая часть (будет доступна бесплатно в личном кабинете Слёрма)

Введение

• Немного истории: почему лог и стриминг важны, и занимают значительное место в бизнес процессах компаний;
• Fun fact: почему "кафка"? А не почему, просто хорошо звучит.

Базовые основы технологии

• Сравнения с подобными технологиями;
• Базовые примитивы: брокеры, топики, партиции, оффсеты, retention.

Установка, запись и чтение создание и конфигурация топиков

• Первая запись и первое чтение;
• Базовые основы Apache Zookeeper.

Продвинутая часть

Тема 1: Работа с распределенным кластером

• Конфигурация кластера,
• Отказоустойчивость,
• Контроллер,
• Репликация между кластерами в разных ЦОД,
• Примеры архитектуры.

Тема 2: Клиентские библиотеки

• Producer,
• Consumer,
• Как (не) потерять данные: консистентность против доступности.

Тема 3: Мониторинг

• Ключевые метрики и алерты,
• SLI & SLO.

Тема 4: Анализ производительности

• Почему Кафка такая быстрая?
• Бенчмаркинг.

Тема 5: Поддержка работоспособности кластера и траблшутинг

• Балансировка нагрузки: partition assignment & repartitioning,
• Обновление версии кластера и клиентов,
• Чтение логов,
• Продвинутый функционал траблшутинга,
• Примеры сбоев из жизни.

Тема 6: Развертывание кластера в проде

• Рекомендуемая конфигурация и архитектура,
• Практики и примеры из жизни.

Авторы курса:

• Анатолий Солдатов, Lead Engineer в Avito;
• Александр Миронов, Infrastructure Engineer в Stripe.

Релиз курса в апреле 2021 года. Сейчас собираем обратную связь по программе, готовим материалы и уже записываем бесплатную базовую часть курса. Оставить заявку на весь курс или бесплатные вводные уроки можно здесь.

Kafka повсюду. Где есть микросервисы и распределенные вычисления, а они сейчас популярны, там почти наверняка есть и Kafka. В статье я попытаюсь объяснить, как в Kafka работает механизм хранения.

Я, конечно, постарался не усложнять, но копать будем глубоко, поэтому какое-то базовое представление о Kafka не помешает. Иначе не все будет понятно. В общем, продолжайте читать на свой страх и риск.

Обычно считается, что Kafka это распределенная и реплицированная очередь сообщений. С технической точки зрения все верно, но термин очередь сообщений не все понимают одинаково. Я предпочитаю другое определение: распределенный и реплицированный журнал коммитов. Эта формулировка кажется более точной, ведь мы все прекрасно знаем, как журналы записываются на диск. Просто в этом случае на диск попадают сообщения, отправленные в Kafka.

Применительно к хранению в Kafka используется два термина: партиции и топики. Партиции это единицы хранения сообщений, а топики что-то вроде контейнеров, в которых эти партиции находятся.

С основной теорией мы определились, давайте перейдем к практике.

Я создам в Kafka топик с тремя партициями. Если хотите повторять за мной, вот как выглядит команда для локальной настройки Kafka в Windows.

kafka-topics.bat --create --topic freblogg --partitions 3 --replication-factor 1 --zookeeper localhost:2181

В каталоге журналов Kafka создано три каталога:

$ tree freblogg*freblogg-0|-- 00000000000000000000.index|-- 00000000000000000000.log|-- 00000000000000000000.timeindex`-- leader-epoch-checkpointfreblogg-1|-- 00000000000000000000.index|-- 00000000000000000000.log|-- 00000000000000000000.timeindex`-- leader-epoch-checkpointfreblogg-2|-- 00000000000000000000.index|-- 00000000000000000000.log|-- 00000000000000000000.timeindex`-- leader-epoch-checkpoint

Мы создали в топике три партиции, и у каждой свой каталог в файловой системе. Еще тут есть несколько файлов (index, log и т д.), но о них чуть позже.

Обратите внимание, что в Kafka топик это логическое объединение, а партиция фактическая единица хранения. То, что физически хранится на диске. Как устроены партиции?

Партиции

В теории партиция это неизменяемая коллекция (или последовательность) сообщений. Мы можем добавлять сообщения в партицию, но не можем удалять. И под мы я подразумеваю продюсеров в Kafka. Продюсер не может удалять сообщения из топика.

Сейчас мы отправим в топик пару сообщений, но сначала обратите внимание на размер файлов в папках партиций.

$ ls -lh freblogg-0total 20M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121   0 Aug  5 08:26 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121   0 Aug  5 08:26 leader-epoch-checkpoint

Как видите, файлы index вместе весят 20 МБ, а файл log совершенно пустой. В папках freblogg-1 и freblogg-2 то же самое.
Давайте отправим сообщения через console producer и посмотрим, что будет:

kafka-console-producer.bat --topic freblogg --broker-list localhost:9092

Я отправил два сообщения сначала ввел стандартное Hello World, а потом нажал на Enter, и это второе сообщение. Еще раз посмотрим на размеры файлов:

$ ls -lh freblogg*freblogg-0:total 20M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121   0 Aug  5 08:26 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121   0 Aug  5 08:26 leader-epoch-checkpointfreblogg-1:total 21M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121  68 Aug  5 10:15 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121  11 Aug  5 10:15 leader-epoch-checkpointfreblogg-2:total 21M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121  79 Aug  5 09:59 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121  11 Aug  5 09:59 leader-epoch-checkpoint

Два сообщения заняли две партиции, и файлы log в них теперь имеют размер. Это потому, что сообщения в партиции хранятся в файле xxxx.log. Давайте заглянем в файл log и убедимся, что сообщение и правда там.

$ cat freblogg-2/*.log@^@^B^@^K^X^@^@^@^A"^@^@^A^VHello World^@

Файлы с форматом log не очень удобно читать, но мы все же видим в конце Hello World, то есть файл обновился, когда мы отправили сообщение в топик. Второе сообщение мы отправили в другую партицию.

Обратите внимание, что первое сообщение попало в третью партицию (freblogg-2), а второе во вторую (freblogg-1). Для первого сообщения Kafka выбирает партицию произвольно, а следующие просто распределяет по кругу (round-robin). Если мы отправим третье сообщение, Kafka запишет его во freblogg-0 и дальше будет придерживаться этого порядка. Мы можем и сами выбирать партицию, указав ключ. Kafka хранит все сообщения с одним ключом в одной и той же партиции.

Каждому новому сообщению в партиции присваивается Id на 1 больше предыдущего. Этот Id еще называют смещением (offset). У первого сообщения смещение 0, у второго 1 и т. д., каждое следующее всегда на 1 больше предыдущего.

<Небольшое отступление>

Давайте используем инструмент Kafka, чтобы понять, что это за странные символы в файле log. Нам они кажутся бессмысленными, но для Kafka это метаданные каждого сообщения в очереди. Выполним команду:

kafka-run-class.bat kafka.tools.DumpLogSegments --deep-iteration --print-data-log --files logs\freblogg-2\00000000000000000000.log

Получим результат:

umping logs\freblogg-2\00000000000000000000.logStarting offset: 0offset: 0 position: 0 CreateTime: 1533443377944 isvalid: true keysize: -1 valuesize: 11 producerId: -1 headerKeys: [] payload: Hello Worldoffset: 1 position: 79 CreateTime: 1533462689974 isvalid: true keysize: -1 valuesize: 6 producerId: -1 headerKeys: [] payload: amazon

(Я удалил из выходных данных кое-что лишнее.)

Здесь мы видим смещение, время создания, размер ключа и значения, а еще само сообщение (payload).

</Небольшое отступление>

Надо понимать, что партиция привязана к брокеру. Если у нас, допустим, три брокера, а папка freblogg-0 существует в broker-1, в других брокерах ее не будет. У одного топика могут быть партиции в нескольких брокерах, но одна партиция всегда существует в одном брокере Kafka (если установлен коэффициент репликации по умолчанию 1, но об этом чуть позже).

Сегменты

Что это за файлы index и log в каталоге партиции? Партиция, может, и единица хранения в Kafka, но не минимальная. Каждая партиция разделена на сегменты, то есть коллекции сообщений. Kafka не хранит все сообщения партиции в одном файле (как в файле лога), а разделяет их на сегменты. Это дает несколько преимуществ. (Разделяй и властвуй, как говорится.)

Главное это упрощает стирание данных. Я уже говорил, что сами мы не можем удалить сообщение из партиции, но Kafka может это сделать на основе политики хранения для топика. Удалить сегмент проще, чем часть файла, особенно когда продюсер отправляет в него данные.

$ ls -lh freblogg-0total 20M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121   0 Aug  5 08:26 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121   0 Aug  5 08:26 leader-epoch-checkpoint

Нули (00000000000000000000) в файлах log и index в каждой папке партиции это имя сегмента. У файла сегмента есть файлы segment.log, segment.index и segment.timeindex.

Kafka всегда записывает сообщения в файлы сегментов в рамках партиции, причем у нас всегда есть активный сегмент для записи. Когда Kafka достигает лимита по размеру сегмента, создается новый файл сегмента, который станет активным.

В имени каждого файла сегмента отражается смещение от первого сообщения. На картинке выше в сегменте 0 содержатся сообщения со смещением от 0 до 2, в сегменте 3 от 3 до 5, и так далее. Последний сегмент, шестой, сейчас активен.

$ ls -lh freblogg*freblogg-0:total 20M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121   0 Aug  5 08:26 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121   0 Aug  5 08:26 leader-epoch-checkpointfreblogg-1:total 21M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121  68 Aug  5 10:15 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121  11 Aug  5 10:15 leader-epoch-checkpointfreblogg-2:total 21M- freblogg 197121 10M Aug  5 08:26 00000000000000000000.index- freblogg 197121  79 Aug  5 09:59 00000000000000000000.log- freblogg 197121 10M Aug  5 08:26 00000000000000000000.timeindex- freblogg 197121  11 Aug  5 09:59 leader-epoch-checkpoint

У нас всего по одному сегменту в каждой партиции, поэтому они называются 00000000000000000000. Раз других файлов сегментов нет, сегмент 00000000000000000000 и будет активным.

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

Допустим, мы отправили в партицию freblogg-2 три сообщения, и она выглядит так:

Три сообщения это наш лимит. На следующем сообщении Kafka автоматически закроет текущий сегмент, создаст новый, сделает его активным и сохранит новое сообщение в файле log этого сегмента. (Я не показываю предыдущие нули, чтобы было проще воспринять).

freblogg-2|-- 00.index|-- 00.log|-- 00.timeindex|-- 03.index|-- 03.log|-- 03.timeindex`--

Удивительное дело, но новый сегмент называется не 01. Мы видим 03.index, 03.log. Почему так?

Kafka называет сегмент по имени минимального смещения в нем. Новое сообщение в партиции имеет смещение 3, поэтому Kafka так и называет новый сегмент. Раз у нас есть сегменты 00 и 03, мы можем быть уверены, что сообщения со смещениями 0, 1 и 2 и правда находятся в сегменте 00. Новые сообщения в партиции freblogg-2 со смещениями 3 ,4 и 5 будут храниться в сегменте 03.

В Kafka мы часто читаем сообщения по определенному смещению. Искать смещение в файле log затратно, особенно если файл разрастается до неприличных размеров (по умолчанию это 1 ГБ). Для этого нам и нужен файл .index. В файле index хранятся смещения и физическое расположение сообщения в файле log.

Файл index для файла log, который я приводил в кратком отступлении, будет выглядеть как-то так:

Если нужно прочитать сообщение со смещением 1, мы ищем его в файле index и видим, что его положение 79. Переходим к положению 79 в файле log и читаем. Это довольно эффективный способ мы быстро находим нужное смещение в уже отсортированном файле index с помощью бинарного поиска.

Параллелизм в партициях

Чтобы гарантировать порядок чтения сообщений из партиции, Kafka дает доступ к партиции только одному консюмеру (из группы консюмеров). Если партиция получает сообщения a, f и k, консюмер читает их в том же порядке: a, f и k. Это важно, ведь порядок потребления сообщений на уровне топика не гарантирован, если у вас несколько партиций.

Если консюмеров будет больше, параллелизм не увеличится. Нужно больше партиций. Чтобы два консюмера параллельно считывали данные из топика, нужно создать две партиции по одной на каждого. Партиции в одном топике могут находиться в разных брокерах, поэтому два консюмера топика могут считывать данные из двух разных брокеров.

Топики

Наконец, переходим к топикам. Мы уже кое-что знаем о них. Главное, что нужно знать: топик это просто логическое объединение нескольких партиций.

Топик может быть распределен по нескольким брокерам через партиции, но каждая партиция находится только в одном брокере. У каждого топика есть уникальное имя, от которого зависят имена партиций.

Репликация

Как работает репликация? Создавая топик в Kafka, мы указываем для него коэффициент репликации replication-factor. Допустим, у нас два брокера и мы устанавливаем replication-factor 2. Теперь Kafka попытается всегда создавать бэкап, или реплику, для каждой партиции в топике. Kafka распределяет партиции примерно так же, как HDFS распределяет блоки данных по нодам.

Допустим, для топика freblogg мы установили коэффициент репликации 2. Мы получим примерно такое распределение партиций:

Даже если реплицированная партиция находится в другом брокере, Kafka не разрешает ее читать, потому что в каждом наборе партиций есть LEADER, то есть лидер, и FOLLOWERS ведомые, которые остаются в резерве. Ведомые периодически синхронизируются с лидером и ждут своего звездного часа. Когда лидер выйдет из строя, один из in-sync ведомых будет выбран новым лидером, и вы будете получать данные из этой партиции.

Лидер и ведомый одной партиции всегда находятся в разных брокерах. Думаю, не нужно объяснять, почему.

Мы подошли к концу этой длинной статьи. Если вы добрались до этого места поздравляю. Теперь вы знаете почти все о хранении данных в Kafka. Давайте повторим, чтобы ничего не забыть.

Итоги

• В Kafka данные хранятся в топиках.
• Топики разделены на партиции.
• Каждая партиция разделена на сегменты.
• У каждого сегмента есть файл log, где хранится само сообщение, и файл index, где хранится позиция сообщения в файле log.
• У одного топика могут быть партиции в разных брокерах, но сама партиция всегда привязана к одному брокеру.
• Реплицированные партиции существуют пассивно. Вы обращаетесь к ним, только если сломался лидер.

От редакции:
Более подробно от работе с Apache Kafka можно узнать на курсе Слёрма. Курс сейчас в разработке. В программе бесплатные базовые уроки и платная продвинутая часть. Оставьте заявку на курс, чтобы получать уведомления.

Ресурсы:

Схема Kafka https://kafka.apache.org/images/kafka_diagram.png

Когда сервисы интегрируются при помощи Kafka очень удобно использовать REST API, как универсальный и стандартный способ обмена сообщениями. При увеличении количества сервисов сложность коммуникаций увеличивается. Для контроля можно и нужно использовать интеграционное тестирование. Такие библиотеки как testcontainers или EmbeddedServer прекрасно помогают организовать такое тестирование. Существуют много примеров для micronaut, Spring Boot и т.д. Но в этих примерах опущены некоторые детали, которые не позволяют с первого раза запустить код. В статье приводятся примеры с подробным описанием и ссылками на код.

Пример

Для простоты можно принять такой REST API.

/runs POST-метод. Инициализирует запрос в канал связи. Принимает данные и возвращает ключ запроса.
/runs/{key}/status GET-метод. По ключу возвращает статус запроса. Может принимать следующие значения: UNKNOWN, RUNNING, DONE.
/runs /{key} GET-метод. По ключу возвращает результат запроса.

Подобный API реализован у livy, хотя и для других задач.

Реализация

Будут использоваться: micronaut, Spring Boot.

micronaut

Контроллер для API.

import io.micronaut.http.annotation.Body;import io.micronaut.http.annotation.Controller;import io.micronaut.http.annotation.Get;import io.micronaut.http.annotation.Post;import io.reactivex.Maybe;import io.reactivex.schedulers.Schedulers;import javax.inject.Inject;import java.util.UUID;@Controller("/runs")public class RunController {    @Inject    RunClient runClient;    @Inject    RunCache runCache;    @Post    public String runs(@Body String body) {        String key = UUID.randomUUID().toString();        runCache.statuses.put(key, RunStatus.RUNNING);        runCache.responses.put(key, "");        runClient.sendRun(key, new Run(key, RunType.REQUEST, "", body));        return key;    }    @Get("/{key}/status")    public Maybe<RunStatus> getRunStatus(String key) {        return Maybe.just(key)                .subscribeOn(Schedulers.io())                .map(it -> runCache.statuses.getOrDefault(it, RunStatus.UNKNOWN));    }    @Get("/{key}")    public Maybe<String> getRunResponse(String key) {        return Maybe.just(key)                .subscribeOn(Schedulers.io())                .map(it -> runCache.responses.getOrDefault(it, ""));    }}

Отправка сообщений в kafka.

import io.micronaut.configuration.kafka.annotation.*;import io.micronaut.messaging.annotation.Body;@KafkaClientpublic interface RunClient {    @Topic("runs")    void sendRun(@KafkaKey String key, @Body Run run);}

Получение сообщений из kafka.

import io.micronaut.configuration.kafka.annotation.*;import io.micronaut.messaging.annotation.Body;import javax.inject.Inject;@KafkaListener(offsetReset = OffsetReset.EARLIEST)public class RunListener {    @Inject    RunCalculator runCalculator;    @Topic("runs")    public void receive(@KafkaKey String key, @Body Run run) {        runCalculator.run(key, run);    }}

Обработка сообщений происходит в RunCalculator. Для тестов используется особая реализация, в которой происходит переброска сообщений.

import io.micronaut.context.annotation.Replaces;import javax.inject.Inject;import javax.inject.Singleton;import java.util.UUID;@Replaces(RunCalculatorImpl.class)@Singletonpublic class RunCalculatorWithWork implements RunCalculator {    @Inject    RunClient runClient;    @Inject    RunCache runCache;    @Override    public void run(String key, Run run) {        if (RunType.REQUEST.equals(run.getType())) {            String runKey = run.getKey();            String newKey = UUID.randomUUID().toString();            String runBody = run.getBody();            runClient.sendRun(newKey, new Run(newKey, RunType.RESPONSE, runKey, runBody + "_calculated"));        } else if (RunType.RESPONSE.equals(run.getType())) {            runCache.statuses.replace(run.getResponseKey(), RunStatus.DONE);            runCache.responses.replace(run.getResponseKey(), run.getBody());        }    }}

Тест.

import io.micronaut.http.HttpRequest;import io.micronaut.http.client.HttpClient;import static org.junit.jupiter.api.Assertions.assertEquals;public abstract class RunBase {    void run(HttpClient client) {        String key = client.toBlocking().retrieve(HttpRequest.POST("/runs", "body"));        RunStatus runStatus = RunStatus.UNKNOWN;        while (runStatus != RunStatus.DONE) {            runStatus = client.toBlocking().retrieve(HttpRequest.GET("/runs/" + key + "/status"), RunStatus.class);            try {                Thread.sleep(500);            } catch (InterruptedException e) {                e.printStackTrace();            }        }        String response = client.toBlocking().retrieve(HttpRequest.GET("/runs/" + key), String.class);        assertEquals("body_calculated", response);    }}

Для использования EmbeddedServer необходимо.

Подключить библиотеки:

testImplementation("org.apache.kafka:kafka-clients:2.6.0:test")testImplementation("org.apache.kafka:kafka_2.12:2.6.0")testImplementation("org.apache.kafka:kafka_2.12:2.6.0:test")

Тест может выглядеть так.

import io.micronaut.context.ApplicationContext;import io.micronaut.http.client.HttpClient;import io.micronaut.runtime.server.EmbeddedServer;import org.junit.jupiter.api.Test;import java.util.HashMap;import java.util.Map;public class RunKeTest extends RunBase {    @Test    void test() {        Map<String, Object> properties = new HashMap<>();        properties.put("kafka.bootstrap.servers", "localhost:9092");        properties.put("kafka.embedded.enabled", "true");        try (EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer.class, properties)) {            ApplicationContext applicationContext = embeddedServer.getApplicationContext();            HttpClient client = applicationContext.createBean(HttpClient.class, embeddedServer.getURI());            run(client);        }    }}

Для использования testcontainers необходимо.

Подключить библиотеки:

implementation("org.testcontainers:kafka:1.14.3")

Тест может выглядеть так.

import io.micronaut.context.ApplicationContext;import io.micronaut.http.client.HttpClient;import io.micronaut.runtime.server.EmbeddedServer;import org.junit.jupiter.api.Test;import org.testcontainers.containers.KafkaContainer;import org.testcontainers.utility.DockerImageName;import java.util.HashMap;import java.util.Map;public class RunTcTest extends RunBase {    @Test    public void test() {        try (KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.3"))) {            kafka.start();            Map<String, Object> properties = new HashMap<>();            properties.put("kafka.bootstrap.servers", kafka.getBootstrapServers());            try (EmbeddedServer embeddedServer = ApplicationContext.run(EmbeddedServer.class, properties)) {                ApplicationContext applicationContext = embeddedServer.getApplicationContext();                HttpClient client = applicationContext.createBean(HttpClient.class, embeddedServer.getURI());                run(client);            }        }    }}

Spring Boot

Контроллер для API.

import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.*;import java.util.UUID;@RestController@RequestMapping("/runs")public class RunController {    @Autowired    private RunClient runClient;    @Autowired    private RunCache runCache;    @PostMapping()    public String runs(@RequestBody String body) {        String key = UUID.randomUUID().toString();        runCache.statuses.put(key, RunStatus.RUNNING);        runCache.responses.put(key, "");        runClient.sendRun(key, new Run(key, RunType.REQUEST, "", body));        return key;    }    @GetMapping("/{key}/status")    public RunStatus getRunStatus(@PathVariable String key) {        return runCache.statuses.getOrDefault(key, RunStatus.UNKNOWN);    }    @GetMapping("/{key}")    public String getRunResponse(@PathVariable String key) {        return runCache.responses.getOrDefault(key, "");    }}

Отправка сообщений в kafka.

import com.fasterxml.jackson.core.JsonProcessingException;import com.fasterxml.jackson.databind.ObjectMapper;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.stereotype.Component;@Componentpublic class RunClient {    @Autowired    private KafkaTemplate<String, String> kafkaTemplate;    @Autowired    private ObjectMapper objectMapper;    public void sendRun(String key, Run run) {        String data = "";        try {            data = objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(run);        } catch (JsonProcessingException e) {            e.printStackTrace();        }        kafkaTemplate.send("runs", key, data);    }}

Получение сообщений из kafka.

import com.fasterxml.jackson.core.JsonProcessingException;import com.fasterxml.jackson.databind.ObjectMapper;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.stereotype.Component;@Componentpublic class RunListener {    @Autowired    private ObjectMapper objectMapper;    @Autowired    private RunCalculator runCalculator;    @KafkaListener(topics = "runs", groupId = "m-group")    public void receive(ConsumerRecord<?, ?> consumerRecord) {        String key = consumerRecord.key().toString();        Run run = null;        try {            run = objectMapper.readValue(consumerRecord.value().toString(), Run.class);        } catch (JsonProcessingException e) {            e.printStackTrace();        }        runCalculator.run(key, run);    }}

Обработка сообщений происходит в RunCalculator. Для тестов используется особая реализация, в которой происходит переброска сообщений.

import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.UUID;@Componentpublic class RunCalculatorWithWork implements RunCalculator {    @Autowired    RunClient runClient;    @Autowired    RunCache runCache;    @Override    public void run(String key, Run run) {        if (RunType.REQUEST.equals(run.getType())) {            String runKey = run.getKey();            String newKey = UUID.randomUUID().toString();            String runBody = run.getBody();            runClient.sendRun(newKey, new Run(newKey, RunType.RESPONSE, runKey, runBody + "_calculated"));        } else if (RunType.RESPONSE.equals(run.getType())) {            runCache.statuses.replace(run.getResponseKey(), RunStatus.DONE);            runCache.responses.replace(run.getResponseKey(), run.getBody());        }    }}

Тест.

import com.fasterxml.jackson.databind.ObjectMapper;import org.springframework.http.MediaType;import org.springframework.test.web.servlet.MockMvc;import org.springframework.test.web.servlet.MvcResult;import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;import static org.junit.jupiter.api.Assertions.assertEquals;import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;public abstract class RunBase {    void run(MockMvc mockMvc, ObjectMapper objectMapper) throws Exception {        MvcResult keyResult = mockMvc.perform(MockMvcRequestBuilders.post("/runs")                .content("body")                .contentType(MediaType.APPLICATION_JSON)                .accept(MediaType.APPLICATION_JSON))                .andExpect(status().isOk())                .andReturn();        String key = keyResult.getResponse().getContentAsString();        RunStatus runStatus = RunStatus.UNKNOWN;        while (runStatus != RunStatus.DONE) {            MvcResult statusResult = mockMvc.perform(MockMvcRequestBuilders.get("/runs/" + key + "/status")                    .contentType(MediaType.APPLICATION_JSON)                    .accept(MediaType.APPLICATION_JSON))                    .andExpect(status().isOk())                    .andReturn();            runStatus = objectMapper.readValue(statusResult.getResponse().getContentAsString(), RunStatus.class);            try {                Thread.sleep(500);            } catch (InterruptedException e) {                e.printStackTrace();            }        }        String response = mockMvc.perform(MockMvcRequestBuilders.get("/runs/" + key)                .contentType(MediaType.APPLICATION_JSON)                .accept(MediaType.APPLICATION_JSON))                .andExpect(status().isOk())                .andReturn().getResponse().getContentAsString();        assertEquals("body_calculated", response);    }}

Для использования EmbeddedServer необходимо.

Подключить библиотеки:

<dependency>    <groupId>org.springframework.kafka</groupId>    <artifactId>spring-kafka</artifactId>    <version>2.5.10.RELEASE</version></dependency><dependency>    <groupId>org.springframework.kafka</groupId>    <artifactId>spring-kafka-test</artifactId>    <version>2.5.10.RELEASE</version>    <scope>test</scope></dependency>

Тест может выглядеть так.

import com.fasterxml.jackson.databind.ObjectMapper;import org.junit.jupiter.api.Test;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.boot.test.context.TestConfiguration;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Import;import org.springframework.kafka.test.context.EmbeddedKafka;import org.springframework.test.web.servlet.MockMvc;@AutoConfigureMockMvc@SpringBootTest@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})@Import(RunKeTest.RunKeTestConfiguration.class)public class RunKeTest extends RunBase {    @Autowired    private MockMvc mockMvc;    @Autowired    private ObjectMapper objectMapper;    @Test    void test() throws Exception {        run(mockMvc, objectMapper);    }    @TestConfiguration    static class RunKeTestConfiguration {        @Autowired        private RunCache runCache;        @Autowired        private RunClient runClient;        @Bean        public RunCalculator runCalculator() {            RunCalculatorWithWork runCalculatorWithWork = new RunCalculatorWithWork();            runCalculatorWithWork.runCache = runCache;            runCalculatorWithWork.runClient = runClient;            return runCalculatorWithWork;        }    }}

Для использования testcontainers необходимо.

Подключить библиотеки:

<dependency>    <groupId>org.testcontainers</groupId>    <artifactId>kafka</artifactId>    <version>1.14.3</version>    <scope>test</scope></dependency>

Тест может выглядеть так.

import com.fasterxml.jackson.databind.ObjectMapper;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.serialization.StringDeserializer;import org.apache.kafka.common.serialization.StringSerializer;import org.junit.ClassRule;import org.junit.jupiter.api.Test;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.boot.test.context.TestConfiguration;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Import;import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;import org.springframework.kafka.core.*;import org.springframework.test.web.servlet.MockMvc;import org.testcontainers.containers.KafkaContainer;import org.testcontainers.utility.DockerImageName;import java.util.HashMap;import java.util.Map;@AutoConfigureMockMvc@SpringBootTest@Import(RunTcTest.RunTcTestConfiguration.class)public class RunTcTest extends RunBase {    @ClassRule    public static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.3"));    static {        kafka.start();    }    @Autowired    private MockMvc mockMvc;    @Autowired    private ObjectMapper objectMapper;    @Test    void test() throws Exception {        run(mockMvc, objectMapper);    }    @TestConfiguration    static class RunTcTestConfiguration {        @Autowired        private RunCache runCache;        @Autowired        private RunClient runClient;        @Bean        ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {            ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();            factory.setConsumerFactory(consumerFactory());            return factory;        }        @Bean        public ConsumerFactory<Integer, String> consumerFactory() {            return new DefaultKafkaConsumerFactory<>(consumerConfigs());        }        @Bean        public Map<String, Object> consumerConfigs() {            Map<String, Object> props = new HashMap<>();            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");            props.put(ConsumerConfig.GROUP_ID_CONFIG, "m-group");            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);            return props;        }        @Bean        public ProducerFactory<String, String> producerFactory() {            Map<String, Object> configProps = new HashMap<>();            configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());            configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);            configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);            return new DefaultKafkaProducerFactory<>(configProps);        }        @Bean        public KafkaTemplate<String, String> kafkaTemplate() {            return new KafkaTemplate<>(producerFactory());        }        @Bean        public RunCalculator runCalculator() {            RunCalculatorWithWork runCalculatorWithWork = new RunCalculatorWithWork();            runCalculatorWithWork.runCache = runCache;            runCalculatorWithWork.runClient = runClient;            return runCalculatorWithWork;        }    }}

Перед всеми тестами необходимо стартовать kafka. Это делается вот таким вот образом:

kafka.start();

Дополнительные свойства для kafka в тестах можно задать в ресурсном файле.

application.yml

spring:  kafka:    consumer:      auto-offset-reset: earliest

Ресурсы и ссылки

Код для micronaut

Код для Spring Boot

PART 1: TESTING KAFKA MICROSERVICES WITH MICRONAUT

Testing Kafka and Spring Boot

Micronaut Kafka

Spring for Apache Kafka

Пятница самое время занимательных историй. Сегодня предлагаем вам послушать доклады DevOps и SQL-направления с конференции ЮMoneyDay. Специалисты расскажут про:

• устройство кластера логов, который позволяет нам понимать, что происходит с платежами и транзакциями (а также в целом с компонентами и сервисами);
• работу дата-инженеров в машинном обучении;
• внедрение и трансформацию CI/CD.

Делимся ценным опытом, чтобы вы не совершали наших ошибок. Надеемся, будет полезно!

Наши грабли залог вашего успеха

Максим Огрызков, старший системный администратор
В докладе пойдет речь об обработке логов нескольких дата-центров с доступом через единый интерфейс. Обсудим причины и последствия обновления кластера. Расскажу о транспорте доставки логов из разных систем и окружений, и причем тут Apache Kafka. А также почему мы не используем logstash и как одним запросом в Kibana приложить кластер.

1:17 О чем будет доклад: кластер логов
1:43 Как логи попадают в кластер?
3:50 Почему мы выбрали Apache Kafka
5:02 Rsyslog: преимущества использования
9:00 Где хранить логи из разных ДЦ?
12:08 Что делать, если объем данных слишком большой?
14:00 Обновление кластера.
20:30 Наши грабли и пути решения
22:35 Translog
24:25 Bulk request
26:28 Opendistro-perfomance-analyzer
28:28 Index Shrink
29:49 Librdkafka
31:37 Итоги: как выглядит наш кластер сейчас

---

Дата-инженеры в машинном обучении

Евгений Виноградов, руководитель отдела разработки хранилищ данных
Рассказ о том, как выглядит промышленная работа над экспериментами в ML какие проблемы решаются на уровне модели, а какие-только на уровне данных, и как обеспечить контролируемый процесс обучения.

1:40 Справка о спикере
2:41 Кто занимается DS-проектами?
8:30 Что такое Data Science проект?
14:15 Порядок действий в DS-проекте
15:42 Процесс сбора датасета
20:26 Как все устроено в Apache Kafka
29:10 Что происходит после сбора датасета
29:21 Как выбрать модель?
30:40 Примеры проблем, которые может решить дата-инженер
34:38 На каких технологиях все это работает?
35:03 Выводы доклада

---

CI/CD для дата-инженера: туда и обратно

Антон Спирин, старший разработчик BI
Доклад о внедрении принципов CI/CD в BI-разработке, целях, их трансформации и преодолении трудностей.

2:00 Справка о спикере
2:44 Описание проблемы
4:28 Кто такой дата-инженер?
5:43 CI/CD в чем состоит работа инженера?
6:55 Подробнее о стеке и информационных системах
8:00 Точка отсчета: с чего мы начинала
10:34 Первый этап изменений
15:50 Кажется, все хорошо, но второй этап улучшений
19:01 Почти демо: JenkinsFile, Pipelines
20:44 Что мы получили на выходе?
22:43 Сколько ушло времени? Статистика по релизам
23:37 Наши челленджи и что можно было бы сделать иначе. Планы на будущее

---

Все доклады с большой ИТ-конференции ЮMoneyDay. На подходе материалы про PM, тестирование и мобильную разработку.

Введение

Существуют различные способы для написания тестов с использованием Apache Kafka. К примеру, можно использовать TestContainers и EmbeddedKafka. Об этом можно почитать, к примеру, вот здесь: Подводные камни тестирования Kafka Streams. Но существует и вариант для написания тестов с использованием KafkaServer.

Что будет тестироваться?

Предположим, необходимо разработать сервис отправки сообщений по различным каналам: email, telegram и т.п.

Пусть имя сервиса будет: SenderService.

Сервис должен: слушать заданный канал, выделять из канала нужные ему сообщения, разбирать сообщения и отправлять по нужному каналу для конечной доставки сообщений.

Для проверки сервиса необходимо сформировать сообщение для отправки с использованием канала отправки почты и убедиться в том, что сообщение было передано в конечный канал.
Конечно, в реальных приложениях тесты будут сложнее. Но для иллюстрации выбранного подхода, такого теста будет достаточно.

Сервис и тест реализованы с использованием: Java 1.8, Kafka 2.1.0, JUnit 5.5.2, Maven 3.6.1.

Сервис

Сервис будет иметь возможность начать работу и остановить свою работу.

void start()void stop()

При старте необходимо задать, как минимум, следующие параметры:

String bootstrapServersString senderTopicEmailService emailService

bootstrapServers адрес kafka.
senderTopic топик, из которого будут считываться сообщения.
emailService сервис для конечной отправки сообщений по почте.

В реальном сервисе таких конечных сервисов будет столько же сколько и конечных каналов отправки сообщений.

Теперь необходим потребитель, который слушает канал, фильтрует и отправляет сообщения в конечные каналы. Количество таких потребителей можно выбирать. Подход для написания потребителя описан вот здесь: Introducing the Kafka Consumer: Getting Started with the New Apache Kafka 0.9 Consumer Client.

Collection<AutoCloseable> closeables = new ArrayList<>();ExecutorService senderTasksExecutor = Executors.newFixedThreadPool(senderTasksN);ExecutorService tasksExecutorService = Executors.newFixedThreadPool(tasksN);for (int i = 0; i < senderTasksN; i++) {    SenderConsumerLoop senderConsumerLoop =            new SenderConsumerLoop(                    bootstrapServers,                    senderTopic,                    "sender",                    "sender",                    tasksExecutorService,                    emailService            );    closeables.add(senderConsumerLoop);    senderTasksExecutor.submit(senderConsumerLoop);}

В цикле создается экземпляр потребителя, запоминается в коллекции и запускается через сервис запуска задач.

При выполнении этого кода потребители начинают работать. Сервис ждет их завершения или сигнала для остановки.

Runtime.getRuntime().addShutdownHook(new Thread(() -> {    for (AutoCloseable autoCloseable : closeables) {        try {            autoCloseable.close();        } catch (Exception e) {            e.printStackTrace();        }    }    senderTasksExecutor.shutdown();    tasksExecutorService.shutdown();    stop();    try {        senderTasksExecutor.awaitTermination(5000, TimeUnit.MILLISECONDS);    } catch (InterruptedException e) {        e.printStackTrace();    }}));

При завершении необходимо освободить ресурсы.

Потребитель

Потребитель имеет следующие публичные методы:

void run()void close()

Основной метод: run.

@Overridepublic void run() {    kafkaConsumer = createKafkaConsumerStringString(bootstrapServers, clientId, groupId);    kafkaConsumer.subscribe(Collections.singleton(topic));    while (true) {        calculate(kafkaConsumer.poll(Duration.ofSeconds(1)));    }}

По входным параметрам создается экземпляр kafka-потребителя. kafka-потребитель подписывается на заданный топик. В бесконечном цикле выбираются записи из топика. И отправляются на обработку.

Для иллюстрации json-сообщения будут иметь несколько полей, которые будут задавать и тип сообщения, и данные для отправки.

Пример сообщения:

{  "subject": {    "subject_type": "send"  },  "body": {    "method": "email",    "recipients": "mrbrown@ml.ml;mrblack@ml.ml;mrwhite@ml.ml",    "title": "42",    "message": "73"  }}

subject_type тип сообщения. Для сервиса нужно значение send.
method тип конечного сервиса для отправки. email отправка через почту.
recipients список получателей.
title заголовок для сообщения.
message сообщение.

Обработка всех записей:

void calculate(ConsumerRecords<String, String> records) {    for (ConsumerRecord<String, String> record : records) {        calculate(record);    }}

Обработка одной записи:

void calculate(ConsumerRecord<String, String> record) {            JSONParser jsonParser = new JSONParser();            Object parsedObject = null;            try {                parsedObject = jsonParser.parse(record.value());            } catch (ParseException e) {                e.printStackTrace();            }            if (parsedObject instanceof JSONObject) {                JSONObject jsonObject = (JSONObject) parsedObject;                JSONObject jsonSubject = (JSONObject) jsonObject.get(SUBJECT);                String subjectType = jsonSubject.get(SUBJECT_TYPE).toString();                if (SEND.equals(subjectType)) {                    JSONObject jsonBody = (JSONObject) jsonObject.get(BODY);                    calculate(jsonBody);                }            }        }

Распределение сообщений по типу:

void calculate(JSONObject jsonBody) {    String method = jsonBody.get(METHOD).toString();    if (EMAIL_METHOD.equals(method)) {        String recipients = jsonBody.get(RECIPIENTS).toString();        String title = jsonBody.get(TITLE).toString();        String message = jsonBody.get(MESSAGE).toString();        sendEmail(recipients, title, message);    }}

Отправка в конечную систему:

void sendEmail(String recipients, String title, String message) {    tasksExecutorService.submit(() -> emailService.send(recipients, title, message));}

Отправка сообщений происходит через сервис исполнения задач.

Ожидания завершения отправки не происходит.

Создание kafka-потребителя:

static KafkaConsumer<String, String> createKafkaConsumerStringString(        String bootstrapServers,        String clientId,        String groupId) {    Properties properties = new Properties();    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);    properties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, clientId);    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);    properties.setProperty(            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,            "org.apache.kafka.common.serialization.StringDeserializer");    properties.setProperty(            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,            "org.apache.kafka.common.serialization.StringDeserializer");    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");    return new KafkaConsumer<>(properties);}

Интерфейс для писем:

interface EmailService {    void send(String recipients, String title, String message);}

Тест

Для теста понадобиться следующее.
Адрес kafka-сервера.
Порт для kafka-сервера.
Имя топика.

Сервис для управления kafka-сервером. Будет описан ниже.

public class SenderServiceTest {    @Test    void consumeEmail() throws InterruptedException {        String brokerHost = "127.0.0.1";        int brokerPort = 29092;        String bootstrapServers = brokerHost + ":" + brokerPort;        String senderTopic = "sender_data";        try (KafkaServerService kafkaServerService = new KafkaServerService(brokerHost, brokerPort)) {            kafkaServerService.start();            kafkaServerService.createTopic(senderTopic);        }    }}

Задаются параметры. Создается сервис для управления kafka-сервером. kafka-сервером стартует. Создается необходимый топик.

Создается mock конечного сервиса для отправки сообщений:

SenderService.EmailService emailService = mock(SenderService.EmailService.class);

Создается сам сервис и стартует:

SenderService senderService = new SenderService(bootstrapServers, senderTopic, emailService);senderService.start();

Задаются параметры для сообщения:

String recipients = "mrbrown@ml.ml;mrblack@ml.ml;mrwhite@ml.ml";String title = "42";String message = "73";

Отправляется сообщение в канал:

kafkaServerService.send(senderTopic, key(), createMessage(EMAIL_METHOD, recipients, title, message));

Ожидание:

Thread.sleep(6000);

Проверка, что сообщение дошло до конечного сервиса:

verify(emailService).send(recipients, title, message);

Остановка:

senderService.stop();

Все вместе:

public class SenderServiceTest {    @Test    void consumeEmail() throws InterruptedException {        String brokerHost = "127.0.0.1";        int brokerPort = 29092;        String bootstrapServers = brokerHost + ":" + brokerPort;        String senderTopic = "sender_data";        try (KafkaServerService kafkaServerService = new KafkaServerService(brokerHost, brokerPort)) {            kafkaServerService.start();            kafkaServerService.createTopic(senderTopic);            SenderService.EmailService emailService = mock(SenderService.EmailService.class);            SenderService senderService = new SenderService(bootstrapServers, senderTopic, emailService);            senderService.start();            String recipients = "mrbrown@ml.ml;mrblack@ml.ml;mrwhite@ml.ml";            String title = "42";            String message = "73";            kafkaServerService.send(senderTopic, key(), createMessage(EMAIL_METHOD, recipients, title, message));            Thread.sleep(6000);            verify(emailService).send(recipients, title, message);            senderService.stop();        }    }}

Вспомогательный код:

public class SenderFactory {    public static final String SUBJECT = "subject";    public static final String SUBJECT_TYPE = "subject_type";    public static final String BODY = "body";    public static final String METHOD = "method";    public static final String EMAIL_METHOD = "email";    public static final String RECIPIENTS = "recipients";    public static final String TITLE = "title";    public static final String MESSAGE = "message";    public static final String SEND = "send";    public static String key() {        return UUID.randomUUID().toString();    }    public static String createMessage(String method, String recipients, String title, String message) {        Map<String, Object> map = new HashMap<>();        Map<String, Object> subject = new HashMap<>();        Map<String, Object> body = new HashMap<>();        map.put(SUBJECT, subject);        subject.put(SUBJECT_TYPE, SEND);        map.put(BODY, body);        body.put(METHOD, method);        body.put(RECIPIENTS, recipients);        body.put(TITLE, title);        body.put(MESSAGE, message);        return JSONObject.toJSONString(map);    }}

Сервис для управления kafka-сервером

Основные методы:

void start()void close()void createTopic(String topic)

В методе start происходит создание сервера и вспомогательных объектов.

Создание zookeeper и сохранение его адреса:

zkServer = new EmbeddedZookeeper();String zkConnect = zkHost + ":" + zkServer.port();

Создание клиента zookeeper:

zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer$.MODULE$);zkUtils = ZkUtils.apply(zkClient, false);

Задание свойств для сервера:

Properties brokerProps = new Properties();brokerProps.setProperty("zookeeper.connect", zkConnect);brokerProps.setProperty("broker.id", "0");try {    brokerProps.setProperty("log.dirs", Files.createTempDirectory("kafka-").toAbsolutePath().toString());} catch (IOException e) {    throw new RuntimeException(e);}brokerProps.setProperty("listeners", "PLAINTEXT://" + brokerHost + ":" + brokerPort);brokerProps.setProperty("offsets.topic.replication.factor", "1");KafkaConfig config = new KafkaConfig(brokerProps);

Создание сервера:

kafkaServer = TestUtils.createServer(config, new MockTime());

Все вместе:

public void start() {    zkServer = new EmbeddedZookeeper();    String zkConnect = zkHost + ":" + zkServer.port();    zkClient = new ZkClient(zkConnect, 30000, 30000, ZKStringSerializer$.MODULE$);    zkUtils = ZkUtils.apply(zkClient, false);    Properties brokerProps = new Properties();    brokerProps.setProperty("zookeeper.connect", zkConnect);    brokerProps.setProperty("broker.id", "0");    try {        brokerProps.setProperty("log.dirs", Files.createTempDirectory("kafka-").toAbsolutePath().toString());    } catch (IOException e) {        throw new RuntimeException(e);    }    brokerProps.setProperty("listeners", "PLAINTEXT://" + brokerHost + ":" + brokerPort);    brokerProps.setProperty("offsets.topic.replication.factor", "1");    KafkaConfig config = new KafkaConfig(brokerProps);    kafkaServer = TestUtils.createServer(config, new MockTime());}

Остановка сервиса:

@Overridepublic void close() {    kafkaServer.shutdown();    zkClient.close();    zkServer.shutdown();}

Создание топика:

public void createTopic(String topic) {    AdminUtils.createTopic(            zkUtils, topic, 1, 1, new Properties(), RackAwareMode.Disabled$.MODULE$);}

Заключение

В заключении нужно отметить, что приведенный здесь код лишь иллюстрирует выбранный способ.

Для создания и тестирования сервисов с использованием kafka можно обратиться к следующему ресурсу:
kafka-streams-examples

Ссылки и ресурсы

Исходный код

Код для тестирования с kafka-сервером

Введение

Во время работы над задачами машинного обучения с онлайн-данными есть необходимость собирать различные сущности в одну для дальнейшего анализа и оценки. Процесс сбора должен быть удобным и быстрым. А также часто должен предусматривать бесшовный переход от процесса разработки к промышленному использованию без дополнительных усилий и рутинной работы. Для решения этой проблемы можно воспользоваться подходом с использованием Feature Store. Этот подход со многими деталями описан вот здесь: Meet Michelangelo: Ubers Machine Learning Platform. В этой статье описывается интерпретация указанного решения для управления признаками в виде прототипа.

Feature Store для онлайн потоков

Feature Store можно рассматривать как сервис, который должен выполнять свои функции строго по его спецификации. Прежде чем определить эту спецификацию, следует разобрать простой пример.

Пример

Пусть даны следующие сущности.
Фильм, который обладает идентификатором и заголовком.
Рейтинг фильма, у которого так же есть собственный идентификатор, идентификатор фильма, а также значение рейтинга. Рейтинг меняется во времени.
Источник рейтинга, который так же имеет собственный рейтинг. И меняется во времени.
И нужно эти сущности объединить в одну.
Вот что получается.

Диаграмма сущностей.

Как можно понять, объединение происходит по ключам сущностей. Т.е. к фильму ищутся все рейтинги фильма, а к рейтингу фильма все рейтинги источника.

Обобщение примера

Теперь можно обобщить пример и масштабировать его на все сущности, которые могут быть связаны по ключам.

Есть kafka-потоки, которые определяют собой сущности: A, B NN.
Нужно объединять эти потоки для создания новых потоков: AB, BCD NM.
Этим процессом должен управлять сервис: Feature Stream Engine.
Feature Stream Engine умеет объединять сущности в kafka-потоках, используя хранилище метаданных Feature Stream Store и Feature Stream Center, как единую точку входа по управлению объединением.

Обобщенная диаграмма сущностей и Feature Stream Engine.

Feature Stream Store

Хранилище метаданных представляет из себя сервис по хранению данных о потоках, сущностях и их связях.

Основная единица хранилища это признак (feature).
Признак имеет свой идентификатор, ссылку на источник, наименование и тип.

Источник группирует признаки и привязывается к определенному потоку.

Feature Stream Center

Центр управления позволяет создавать новые потоки, а также взаимодействовать со службами доставки и развертывания для поддержки работы новых потоков в различных средах исполнения, в том числе и промышленной среде.

Feature Stream Engine

Feature Stream Engine обеспечивает работу с потоками, а так же взаимодействие с внешними сервисами и конечными пользователями.

Компоненты Feature Stream Engine.

Архитектура Feature Stream Engine

Feature Stream Engine представляет из себя конструктор, который позволяет собирать признаки из различных потоков и доставлять этот функционал на различные среды.

Feature Stream Engine должен реализовывать следующие функции.

Описывать источники данных.
Привязывать источники данных к потокам kafka.
Описывать признаки и привязывать их к источникам данных.
Создавать новые источники данных на основе имеющихся путем объединения по ключам (особым признакам).
Развертывать функционал работы потоков в различных средах, включая промышленную среду.

Архитектура Feature Stream Engine.

Прототип

Для реализации идеи необходимо упросить функциональность.

Будут объединяться несколько потоков по ключам и записываться в один поток.

Предположим, что метаданные описываются файлами со свойствами ("configration.properties").

Эти данные реализуют следующею модель.

Источники данных в виде имен topic-ов для kafka. Перечисляются через ,.
Ключи в этих источниках данных. Перечисляются через ,.
Имя результирующего topic-а.

Конвертация входных параметров в структуру, которая описывает объединение потоков.

public static FeaturesDescriptor createFromProperties(Properties properties) {    String sources = properties.getProperty(FEATURES_DESCRIPTOR_FEATURE_DESCRIPTORS_SOURCES);    String keys = properties.getProperty(FEATURES_DESCRIPTOR_FEATURE_DESCRIPTORS_KEYS);    String sinkSource = properties.getProperty(FEATURES_DESCRIPTOR_SINK_SOURCE);    String[] sourcesArray = sources.split(",");    String[] keysArray = keys.split(",");    List<FeatureDescriptor> featureDescriptors = new ArrayList<>();    for (int i = 0; i < sourcesArray.length; i++) {        FeatureDescriptor featureDescriptor =                new FeatureDescriptor(sourcesArray[i], keysArray[i]);        featureDescriptors.add(featureDescriptor);    }    return new FeaturesDescriptor(featureDescriptors, sinkSource);}

public static class FeatureDescriptor {    public final String source;    public final String key;    public FeatureDescriptor(String source, String key) {        this.source = source;        this.key = key;    }}

public static class FeaturesDescriptor {    public final List<FeatureDescriptor> featureDescriptors;    public final String sinkSource;    public FeaturesDescriptor(List<FeatureDescriptor> featureDescriptors, String sinkSource) {        this.featureDescriptors = featureDescriptors;        this.sinkSource = sinkSource;    }}

Основной метод по объединению.

void buildStreams(StreamsBuilder builder)

Создаются topic-и, в которых ключам являются, те ключи, по которым нужно объединять.

Serde<String> stringSerde = Serdes.String();List<KStream<String, String>> streams = new ArrayList<>();for (FeatureDescriptor featureDescriptor : featuresDescriptor.featureDescriptors) {    KStream<String, String> stream =            builder.stream(featureDescriptor.source, Consumed.with(stringSerde, stringSerde))                    .map(new KeyValueMapperSimple(featureDescriptor.key));    streams.add(stream);}

Само объединение.

KStream<String, String> pref = streams.get(0);for (int i = 1; i < streams.size(); i++) {    KStream<String, String> cur = streams.get(i);    pref = pref.leftJoin(cur,            new ValueJoinerSimple(),            JoinWindows.of(Duration.ofSeconds(1)),            StreamJoined.with(                    Serdes.String(),                    Serdes.String(),                    Serdes.String())    );}

Отправка в конечный topic.

pref.to(featuresDescriptor.sinkSource);

Все вместе.

public void buildStreams(StreamsBuilder builder) {    Serde<String> stringSerde = Serdes.String();    List<KStream<String, String>> streams = new ArrayList<>();    for (FeatureDescriptor featureDescriptor : featuresDescriptor.featureDescriptors) {        KStream<String, String> stream =                builder.stream(featureDescriptor.source, Consumed.with(stringSerde, stringSerde))                        .map(new KeyValueMapperSimple(featureDescriptor.key));        streams.add(stream);    }    if (streams.size() > 0) {        if (streams.size() == 1) {            KStream<String, String> stream = streams.get(0);            stream.to(featuresDescriptor.sinkSource);        } else {            KStream<String, String> pref = streams.get(0);            for (int i = 1; i < streams.size(); i++) {                KStream<String, String> cur = streams.get(i);                pref = pref.leftJoin(cur,                        new ValueJoinerSimple(),                        JoinWindows.of(Duration.ofSeconds(1)),                        StreamJoined.with(                                Serdes.String(),                                Serdes.String(),                                Serdes.String())                );            }            pref.to(featuresDescriptor.sinkSource);        }    }}

Запуск.

void run(Properties config)

Конструируется объект объединения потоков (основной объект).

FeaturesStream featuresStream = new FeaturesStream(config);

Создается обвязка для kafka.

StreamsBuilder builder = new StreamsBuilder();featuresStream.buildStreams(builder);Topology topology = builder.build();KafkaStreams streams = new KafkaStreams(topology, featuresStream.buildStreamsProperties());

Осуществляется запуск.

streams.start();

Все вместе.

public static void run(Properties config) {    StreamsBuilder builder = new StreamsBuilder();    FeaturesStream featuresStream = new FeaturesStream(config);    featuresStream.buildStreams(builder);    Topology topology = builder.build();    KafkaStreams streams = new KafkaStreams(topology, featuresStream.buildStreamsProperties());    CountDownLatch latch = new CountDownLatch(1);    Runtime.getRuntime().addShutdownHook(new Thread(() -> {        streams.close();        latch.countDown();    }));    try {        streams.start();        latch.await();    } catch (Throwable e) {        System.exit(1);    }    System.exit(0);}

Пример запуска приложения.

java -jar features-stream-1.0.0.jar -c plain.properties

Язык: Java 1.8.
Библиотеки: kafka 2.6.0, jsoup 1.13.1.

Заключение

Изложенное решение имеет ряд ограничений и не реализует полный функционал. Но имеет и несколько преимуществ.
Во-первых: позволяет быстро конструировать объединение topic-в.
Во-вторых: позволяет быстро запускать объединение в различных средах.

Стоит отметить, что решение налагает ограничение на структуру входных данных. А именно, topic-и должны иметь табличную структуру. Для преодоления этого ограничения можно ввести дополнительный слой, который будет позволять сводить различные структуры к табличной.

Для промышленной реализации полной функциональности стоит обратить внимание на очень мощный и, самое главное, гибкий функционал: KSQL.

Ссылки и ресурсы

Исходный код;
Meet Michelangelo: Ubers Machine Learning Platform.

Привет, Хабр!

Нас давно занимала тема использования Apache Kafka в качестве хранилища данных, рассмотренная с теоретической точки зрения, например, здесь. Тем интереснее предложить вашему вниманию перевод материала из блога Twitter (оригинал декабрь 2020), в котором описан нетрадиционный вариант использования Kafka в качестве базы данных для обработки и воспроизведения событий. Надеемся, статья будет интересна и натолкнет вас на свежие мысли и решения при работе с Kafka.

Введение

Когда разработчики потребляют общедоступные данные Twitter через API Twitter, они рассчитывают на надежность, скорость и стабильность. Поэтому некоторое время назад в Twitter запустили Account Activity Replay API для Account Activity API, чтобы разработчикам было проще обеспечить стабильность своих систем. Account Activity Replay API это инструмент восстановления данных, позволяющий разработчикам извлекать события со сроком давности до пяти дней. Этот API восстанавливает события, которые не были доставлены по разным причинам, в том числе, из-за аварийных отказов сервера, происходивших при попытках доставки в режиме реального времени.

Инженеры Twitter стремились не только создать API, которые будут положительно восприняты разработчиками, но и:

• Повысить производительность труда инженеров;
• Добиться, чтобы систему было легко поддерживать. В частности, минимизировать необходимость переключения контекстов для разработчиков, SRE-инженеров и всех прочих, кто имеет дело с системой.

По этой причине при работе над созданием системы воспроизведения, на которую полагается при работе API, было решено взять за основу имеющуюся систему для работы в режиме реального времени, на которой основан Account Activity API. Так удалось повторно воспользоваться имеющимися разработками и минимизировать переключение контекста и объем обучения, которые были бы гораздо значительнее, если бы для описанной работы создавалась совершенно новая система.

Решение для работы в режиме реального времени базируется на архитектуре публикация-подписка. Для такой цели, учитывая поставленные задачи и создавая уровень хранения информации, с которого она будет считываться, возникла идея переосмыслить известную потоковую технологию Apache Kafka.

Контекст

События, происходящие в режиме реального времени, производятся в двух датацентрах. Когда эти события порождаются, их записывают в топики, работающие по принципу публикация-подписка, которые в целях обеспечения избыточности подвергаются перекрестной репликации в двух датацентрах.

Доставлять требуется не все события, поэтому все события фильтруются внутренним приложением, которое потребляет события из нужных топиков, сверяет каждое с набором правил в хранилище ключей и значений, и решает, должно ли событие быть доставлено конкретному разработчику через общедоступный API. События доставляются через вебхук, а URL каждого вебхука принадлежит разработчику, идентифицируемому уникальным ID.



Рис. 1: Конвейер генерации данных

Хранение и сегментирование

Как правило, при создании системы воспроизведения, требующей подобного хранилища данных, выбирается архитектура на основе Hadoop и HDFS. В данном случае, напротив, был выбран Apache Kafka, по двум причинам:

• Система для работы в режиме реального времени была по принципу публикация-подписка, органичному для устройства Kafka
• Объем событий, который требуется хранить в системе воспроизведения, исчисляется не петабайтами. Мы храним данные не более чем за несколько дней. Кроме того, обращаться с заданиями MapReduce для Hadoop затратнее и медленнее, чем потреблять данные в Kafka, и первый вариант не отвечает ожиданиям разработчиков.

В данном случае основная нагрузка ложится на конвейер воспроизведения данных, работающий в режиме реального времени и гарантирующий, что события, которые должны быть доставлены каждому разработчику, будут храниться в Kafka. Назовем топик Kafka delivery_log; по одному такому топику будет на каждый датацентр. Эти топики подвергаются перекрестной репликации для обеспечения избыточности, что позволяет выдавать из одного датацентра запрос на репликацию. Перед доставкой сохраненные таким образом события проходят дедупликацию.

В данном топике Kafka мы создаем множество партиций, использующих задаваемое по умолчанию семантическое сегментирование. Следовательно, партиции соответствуют хешу webhookId разработчика, и этот id служит ключом каждой записи. Предполагалось использовать статическое сегментирование, но в итоге от него отказались из-за повышенного риска, что в одной партиции окажется больше данных, чем в других, если какие-то разработчики в ходе своей деятельности генерируют больше событий, чем другие. Вместо этого было выбрано фиксированное количество партиций, по которым распределяются данные, а стратегия сегментирования была оставлена по умолчанию. Таким образом снижается риск возникновения несбалансированных партиций, а также не приходится считывать все партиции в топике Kafka.

Напротив, на основе webhookId, для которого поступает запрос, сервис воспроизведения определяет конкретную партицию, с которой будет происходить считывание, и поднимает новый потребитель Kafka для данной партиции. Количество партиций в топике не меняется, так как от этого зависит хеширование ключей и распределение событий.

Чтобы минимизировать пространство под хранилище данных, информация подвергается сжатию при помощи алгоритма snappy, поскольку известно, что большая часть информации в описываемой задаче обрабатывается на стороне потребителя. Кроме того, snappy быстрее поддается декомпрессии, нежели другие алгоритмы сжатия, поддерживаемые Kafka: gzip и lz4.

Запросы и обработка

В системе, сконструированной таким образом, API отправляет запросы на воспроизведение. В составе полезной нагрузки каждого валидируемого запроса приходит webhookId и диапазон данных, для которых должны быть воспроизведены события. Эти запросы долговременно хранятся в MySQL и ставятся в очередь до тех пор, пока их не подхватит сервис воспроизведения. Диапазон данных, указанный в запросе, используется, чтобы определить смещение, с которым нужно начать считывание с диска. Функция offsetForTimes объекта Consumer используется для получения смещений.



Рис. 2: Система воспроизведения. Она получает запрос и отправляет его сервису конфигурации (уровень доступа к данным) для дальнейшего долговременного хранения в базе данных.

Инстансы сервиса воспроизведения обрабатывают каждый запрос на воспроизведение. Инстансы координируются друг с другом при помощи MySQL, чтобы обработать следующую запись на воспроизведение, хранимую в базе данных. Каждый рабочий процесс, обеспечивающий воспроизведение, периодически опрашивает MySQL, чтобы узнать, нет ли там задания, которое следует обработать. Запрос переходит от состояния к состоянию. Запрос, который не был подхвачен для обработки, находится в состоянии OPEN. Запрос, который только что был выведен из очереди, находится в состоянии STARTED. Запрос, обрабатываемый в данный момент, находится в состоянии ONGOING. Запрос, претерпевший все переходы, оказывается в состоянии COMPLETED. Рабочий процесс, отвечающий за воспроизведение, подбирает только такие запросы, обработка которых еще не началась (то есть, находящиеся в состоянии OPEN).

Периодически, после того, как рабочий процесс выведет запрос из очереди на обработку, он отстукивается в таблице MySQL, оставляя метки времени и демонстрируя тем самым, что задание на воспроизведение до сих пор обрабатывается. В случаях, когда инстанс воспроизводящего рабочего процесса отмирает, не успев закончить обработку запроса, такие задания перезапускаются. Следовательно, воспроизводящие процессы выводят из очереди не только запросы, находящиеся в состоянии OPEN, но и подбирают те запросы, что были переведены в состояние STARTED или ONGOING, но не получили отстука в базе данных, спустя заданное количество минут.



Рис. 3: Уровень доставки данных: сервис воспроизведения опрашивает MySQL по поводу нового задания на обработку запроса, потребляет запрос из топика Kafka и доставляет события через сервис Webhook.

В конечном итоге события из топика проходят дедупликацию в процессе считывания, а затем публикуются по URL вебхука конкретного пользователя. Дедупликация выполняется путем ведения кэша считываемых событий, которые при этом хешируются. Если попадется событие с хешем, идентичный которому уже имеется в хэше, то оно доставлено не будет.
В целом, такое использование Kafka нельзя назвать традиционным. Но в рамках описанной системы Kafka успешно работает в качестве хранилища данных и участвует в работе API, который способствует как удобству работы, так и легкости доступа к данным при восстановлении событий. Сильные стороны, имеющиеся в системе для работы в режиме реального времени, пригодились и в рамках такого решения. Кроме того, темпы восстановления данных в такой системе полностью оправдывают ожидания разработчиков.