Блог компании леруа мерлен

В 2020 мы перезапустили мобильное приложение Леруа Мерлен и добавили в него Kotlin Multiplatform, Jetpack Navigation и многое другое. Сейчас наше приложение самое популярное среди DIY (Do It Yourself). На пути запуска мы столкнулись с рядом сложностей и получили классный опыт, которым хотим поделиться.

Ведущий Алексей Гладков, технический архитектор Леруа Мерлен.

Специальный гость Катя Петрова из Jetbrains! Расскажет о трендах и планах развития KMM.

Описание

19:10-19:50 Live coding

Мурагер Жаилхан, разработчик Леруа Мерлен

Пишем сложную навигацию на jetpack Navigation

19:50-20:30 Live coding

Вячеслав Корниенко, разработчик Леруа Мерлен

Покрываем тестами фичу в KMM

20:3020:50 regular talk

Катя Петрова, DevAdvocate, JetBrains

Что произошло после выхода в альфу, что нам принесет релиз 1.5 и что нас ждет в будущем.

Когда: 22 апреля, 19:0021:00 мск, Четверг
Ссылка на мероприятие: https://leroy-merlin.timepad.ru/event/1607225/

История

Некоторое время назад IT в Леруа Мерлен претерпело довольно много изменений. Было перепилено довольно много систем, практически всё писались с нуля. Одной из таких систем была Публикационная Платформа, которая занималась передачей данных о продуктах на сайт и пришла на замену старой системе OPUS. Помимо всего прочего, в этой системе были микросервисы, отвечающие за рекомендации. Я не буду тратить время читателя на объяснение старой логики наполнения рекомендаций, скажу лишь, что первую половину 2020 года на официальном сайте крупнейшего ритейлера можно было увидеть следующее.

Как замену для

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

А к мусорному мешку предлагалось прикупить защитный комбинезон и респиратор, и это ещё до пандемии)

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

Проблематика

У нас есть много специалистов, которые любят свою работу и хорошо знают свой товар. Они на 100% знают, что можно предложить в качестве замены, если искомого товара нет в наличии
(похожие товары) или что ещё может понадобиться дополнительно (сопутствующие товары). Большинство этих сотрудников работает в магазинах и ежедневно консультирует наших клиентов как раз по этим вопросам, но мы не можем попросить их составлять для нас рекомендации: им есть чем заняться. Специфика ритейла заключается в том, что за год меняется до 20% ассортимента. А товаров очень много. Вбивать всё это руками не вариант, нужен другой подход.

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

как из массива чеков получить информацию о том, какой товар к какому подходит?

Word2Vec

Алгоритм, который мы использовали, был разработан Google в 2013 году и называется Word2Vec.
Как следует из его названия, он переводит слово в вектор на основании контекста, на котором его обучают. При этом после обучения схожие по контексту слова будут иметь близкие векторы, а различные далёкие.

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

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

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

Если же мы добавим к контексту скорость, то получим совершенно другую картину:

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

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

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

Word2Vec в рекомендациях

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

Возникает вопрос: как в этих чеках наша модель сможет найти какие-то закономерности?
Можно представить себе следующую аналогию. Допустим, есть 2 чека:

В этих чеках все товары одинаковые, кроме 13227536 и 33444222. Что это означает?
С некой долей вероятности мы можем предположить, что было 2 клиента, которые решали одну и ту же задачу (т. к. все, кроме одного, товары одинаковые), и они её решили, но решили по-разному. Т. е. с некой долей вероятности можно предположить, что товар 13227536 может быть заменён на 33444222, т. е. это похожие товары.

Теперь рассмотрим другой случай:

Тут чек номер 2 включает в себя чек номер 1. С некой долей вероятности мы можем предположить, что также были 2 клиента, решавшие одну задачу, но клиент, оформивший второй чек, решил купить больше товаров. Тут мы можем предположить, что товары, отмеченные фиолетовым, 12355478 78856433 являются сопутствующими товарами к товарам, которые отмечены жёлтым.

Теперь перейдём к обещанным двум строчкам кода.

Код

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

model = gensim.models.Word2Vec( rаw_сhecks, size=200, window=100, min_count=10, workers=4 )

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

similars = model.similar_by_word( product_id, 2000 )

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

substitutes = filter_same_type( similars, product_id_type )complements = filter_different_type( similars, product_id_type )

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

Итоги

А/Б-тест движка для похожих товаров показал общий прирост конверсии аж на 10% и на следующий же день был установлен в продакшн на 100% трафика. Хотелось бы заметить, что это не потому, что подход какой-то революционный, просто скорее всего то, что было раньше, никуда не годилось.

А/Б-тест для сопутствующих товаров показал прирост конверсии только в определённых категориях. И мы знаем почему: наши сотрудники часть сопутствующих товаров выставляют руками, а живая экспертиза довольно часто оказывается лучше машинного обучения, особенно если алгоритм такой незамысловатый. Тем не менее практика показывает, что мы можем рассматривать машинное обучение как помощник для людей, например, там, где у коллег не хватает ресурсов.

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

Минусы подхода

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

Из минусов подхода можно перечислить:

• Word2Vec вряд ли подойдёт для рекомендаций фильмов, музыки, книг и т. д. Это не про предпочтение определённого автора или жанра. В специфике DIY товары покупаются комплектом и под проект: купил обои тебе точно нужен клей. Необходимость в клее не исчезнет, а вот боевики могут в какой-то период разонравиться, и хорошая рекомендательная система должна это как можно раньше диагностировать.

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

• Голый Word2Vec в данной статье учитывает только историю продаж. И не учитывает ничего больше. Это плохо. Например, существует много других способов находить похожие товары по схожести фотографий, похожему названию и т. д.

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

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

Типовая архитектура оркестратора

Классическая схема оркестратора для ритейла выглядит примерно так:

Фронт запрашивает id одного или нескольких товаров, которые хочет отобразить + id маски. Маска - это alias для списка атрибутов, необходимых для отображения. На разных страницах сайта необходимо показывать разный набор атрибутов, информация об этом лежит в отдельном микросервисе. Далее, имея список товаров и необходимых атрибутов, оркестратор идёт в микросервисы, которые информацией по этим атрибутам обладают. Плюс, надо зайти в рекомендательную систему, которая вернёт список id товаров которые мы можем порекомендовать купить вместе с основным (заменители основного и ему сопутствующие). Этот список тоже нужно обогатить атрибутами по маске, которую запросит фронт.

В общем, логика не бог весть какая, но она есть.

Варианты оптимизации

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

Прикрутить кэш

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

Распараллелить запросы

Как показывает практика, большинство времени подобного рода сервисы просто ждут ответа от их доменных коллег, так что если они могут "ждать" эти ответы не по очереди, а одновременно, то хорошая идея это организовать. Если мы берём стандартный сервер tomcat или undertow, то можно сделать так, чтобы каждый запрос, принимаемый сервером порождал N потоков, каждый из которых ходил бы за своим атрибутом, а поток-родитель потом это всё бы собирал и выплёвывал наверх.

Соединить запросы

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

Что имеется в виду: в сервис приходит массив основных продуктов и их атрибутов, по которым нужно собрать информацию. Помимо этого, к продуктам нужно подобрать рекомендации. Сервис рекомендаций, естественно, возвращает только список продуктовых id, которые также нужно обогатить своим набором атрибутов. Атрибуты эти, отличаются от атрибутов основного продукта и тоже приходят как параметр запроса нашего оркестратора.
Фактически, в оркестратор приходит список основных продуктов и 2 маски: одна для основных запрашиваемых продуктов, и вторая для продуктов, которые мы к ним рекомендуем:

{  "products": ["product1", "product2", ..., "productN"],  "mask" : "product_detail_page_attributes",  "recommendation_mask": "name_and_photo_only"}

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

Вариант 1:

1. Идём с известными основными продуктами и их атрибутами по микросервисам собирать о них информацию;

2. Параллельно с этим, идём в сервис рекомендаций, который возвращает нам список id рекомендуемых продуктов;

3. После того, как сервис рекомендаций вернул список товаров, аналогично первому пункту, идём с уже другим набором атрибутов по сервисам и получаем всю необходимую информацию.

Тут 1 и 2 выполняются параллельно, 3 же придётся подождать первых двух. Как видно, число последовательных запросов: 2, что не так много, но, как показывает практика, проблема может заключаться в их количестве. Это то, что у нас было с самого начала, после этого мы сделали следующую оптимизацию:

Вариант 2:

1. Сначала идём в сервис рекомендаций и получаем id продуктов-рекомендаций;

2. После этого добавляем к основным продуктам рекомендованные товары, и к атрибутам основных товаров атрибуты товаров-рекомендаций, и уже со всем этим добром идём к доменным сервисам.

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

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

Тем не менее, второй вариант всё же по моему мнению лучше, т.к.
меньше запросов => меньше потоков => сервис больше таких запросов сможет обработать. Также, доп запрос - это доп затраты на маршаллинг и демаршаллинг, которые могут быть значительными.

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

Выводы:

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

Едем дальше

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

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

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

• Не всегда возможно из-за ограничений инфраструктуры. Это сейчас у нас кубер и со скалированием проблем нет, но перед пандемией у нас был всего десяток серверов для приложений и на одном сервере мы мы могли поднять только 1 инстанс оркестратора. Т.е. максимальное количество инстансов, которые мы могли себе позволить - 10;

• Как показывает практика, это не оптимальный подход. Да, прирост производительности есть, но до для того, чтобы получить необходимые 60ms на 95-й персентиле нужно было этими инстансами прямо таки обмазаться, чего мы себе позволить не могли.

  Время ответа при 5-ти и 8-ми инстансах оркестратора

Анализ проблемы

На самом деле проблему "как разогнать сервис до определённого rps?" можно переформулировать в вопрос "почему на нужном rps сервис не держит?". Для того, чтобы понять в чём проблема, даём нарастающую нагрузку, и, в момент когда сервису плохеет смотрим на состояние.
К сожалению, идея написать статью появилась после того, как мы решили проблему, по этому графики не сохранились, но, если в двух словах, мы увидели, что количество потоков постоянно росло и в какой-то момент garbage collector просто сходит с ума. Причём, ни увеличение возможного количества потоков, ни добавление памяти картину не меняло.

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

1. Сервер принимает входящее http соединение. Занимаются этим потоки пула WORKER_IO_THREADS, их обычно по количеству ядер (в нашем примере 4). Допустим, наш запрос попал на XNIO-1 I/O 2. Он установил соединение, и отправил запрос дальше, освобождаясь для принятия новых запросов.

2. От потока из пула WORKER_IO_THREADS запрос попадает уже на непосредственный обработчик - в нашем случае это XNIO-1 task-2 из пула WORKER_TASK_CORE_THREADS
   Это тот поток, который будет выполнять всё, что мы напишем внутри нашего контроллера.

3. Как уже говорилось выше, зная список атрибутов, XNIO-1 task-2 создаёт n потоков, каждый из которых параллельно должен добыть какую-то часть информации о продуктах. Потоки берутся из кастомного пула (пулы WORKER_IO_THREADS и WORKER_TASK_CORE_THREADS - это сущности undertow)

4. Каждый кастомный поток, идёт к клиенту доменного сервиса и делает запрос. Т.к. под капотом используется RestTemplate, поток блокируется и ждёт пока доменный сервис не ответит.

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

Переход на неблокирующий подход

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

Чтобы понять в чём суть, мне бы хотелось привести аналогию с фастфудом

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

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

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

Вернёмся к нашему серверу

Если используется блокирующий RestTemplate, то поток, который получает информацию от доменного сервиса делает следующее:
Он стучится к сервису, говорит ему "здрасьте", я такое-то приложение, мне нужна такая-то информация, я буду ждать от вас ответ на таком-то порту, и начинает ждать, что на этот порт от сервиса придёт ответ. Пока ответ не придёт, или пока не наступит таймаут поток ничем полезным заниматься не будет.

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

Потоки из этого нового пула будут принимать "заказы" на общение с сервисами моментально возвращать управление вместе с каким-то аналогом Future и избавляя от необходимости заказчика ждать.

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

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

Теория - это классно, как применить-то?

C 11-й Java доступен httpClient, который реализует описанный выше неблокирующий подход. А с Spring 5 есть его обёртка - WebClient, которым можно заменить RestTemplate.
WebClient - часть spring-webflux, по этому как результат запроса в доменный сервис возвращает Mono/Flux. Наш основной же код приложения работает с CompletableFuture.

Возникает вопрос: можно ли использовать WebClient как замену RestTemplate, получив преимущества неблокирующего приложения, но при этом не переписывая с CompletableFuture на Mono/Flux весь проект?

Ответ: можно, но нужно очень хорошо понимать что делаешь.

Как положить прод, если не любишь читать документацию.

У Mono/Flux есть метод toFuture(), который возвращает как раз CompletableFuture.

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

Потоки WebClient(по умолчанию они зовутся reactor-http-epoll-число), которые должны были постоянно работать оббегая каждый свой список портов, заблокированы.
Всё дело оказалось в методе CompletableFuture thenApply(work) который встретился у нас выше по стеку.
Этот метод, "докидывает" работы в текущий CompletableFuture на котором вызывается. И всё бы ничего, но у нас так получилось, что эта "работа" оказалась блокирующей.

Получилось следующее: через thenApplyпотоку reactor-http-epoll-1 из WebClient было "накинуто" задание сходить в ещё один сервис. Пока он этого не сделает и не вернёт управление наверх он не может выполнять свою основную работу - смотреть не ответили ли доменные сервисы. И причём, в конце этого задания у стояла блокировка, т.е. нужно было обязательно дождаться от этого сервиса ответа. И так исторически сложилось, что библиотека WebClient, распределяющая работу типа "сходить в ещё один сервис" выбрала именно reactor-http-epoll-1 для организации взаимодействия в этом запросе. То есть получилась ситуация, когда поток заблокировался, потому что ждал ответа от сервиса, в то время как он же и должен был этот ответ в получить. Но он не мог бегать и смотреть что ему пришло, т.к. был заблокирован. Вот такой вот deadlock.

Решение в данном случае: использовать thenApplyAsync, чтобы дополнительная работа выполнялась не в потоке WebClient .

Мораль же следующая:

Никогда не позволяйте потокам WebClient блокироваться - они постоянно должны работать.

Именно потому что так просто выстрелить себе в ногу, я бы рекомендовал воздержаться от использования CompletableFuture и Mono/Fluxв одном проекте. Если используете Mono/Flux , то лучше перепишите всё на Netty, благо у CompletableFuture и Mono/Fluxочень похожее API. Вот тут в документации написано почему использовать метод toFuture() конвертирующий одно в другое - не очень хорошая практика.

Хэппи-энд

Тем не менее, после правок, всё завелось. После деплоя видим следующую картину:

Где-то на 17:26 произошёл деплой новой версии. Из графика видно что:

• количество потоков сократилось больше чем в 4 раза (это сократилось количество потоков в CUSTOM_THREAD_POOL с 1000+ до 4-х);

• потоки в статусе timed-waiting исчезли как класс;

• количество потоков в статусе waiting не изменилось, но это понятно - нас всё ещё есть нативные воркеры undertow. Можно использовать Netty, тогда не было бы и их - у приложения тогда бы вообще было бы всего 4 универсальных потока, которые занимались бы всем.

Результаты по перфомансу превзошли все ожидания: 0 ошибок и 80 ms 75-й персентиль, при всего 4-х инстансах приложения.

Необходимая производительность в 60 ms на 95 персентиле достигается путём поднятия единственного дополнительного инстанса.

Выводы

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

Общие рекомендации по проектированию оркестраторов могут быть следующие:
Для начала следует задуматься о:

• кэшировании;

• параллельном выполнении запросов;

• объединении запросов.

Именно эти шаги могут уменьшить время ответа запроса.

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

Не советую использовать CompletableFuture и Mono/Flux в одном проекте. Анализ описанных в данной статье проблем при их появлении не тривиален. Как минимум потому, что их не так просто воспроизвести на тестовых стендах. А если используете, то используйте с умом.

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

В чём минусы подхода?

Обсуждая неблокирующее программирование, мы должны понимать, что идея подобной организации потоков не нова. Может быть, она нова в Java, но в Node.js с самого начала всё так и работало. И тем не менее, когда меня пригласили в Леруа Мерлен распиливать монолит, у которого были проблемы с производительностью, то, что мы распиливали, было как раз Node.js приложением. Это говорит о том, что данный подход не является панацеей. Он хорошо работает, когда потоки не нагружены лишней работой и плохо, когда есть много тяжелых вычислений. Но это уже следующий шаг после внедрения неблокирующего клиента).

Люди строили highload на Java и до jdk11/Spring 5, неблокирующее программирование - это всего лишь один из подходов к организации, который хорошо бы иметь в своём распоряжении и применять когда в нём возникнет необходимость.

Спасибо за внимание!

Всем привет!

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

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

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

Общая схема доставки данных из источников в ODS-слой Greenplum посредством разработанного нами frameworkа приведена ниже:

1. Данные из систем-источников пишутся в Kafka в AVRO-формате, обрабатываются в режиме реального времени Apache NiFi, который сохраняет их в формате parquet на S3.

2. Затем эти файлы с сырыми данными с помощью Sparkа обрабатываются в два этапа:

   1. Compaction на данном этапе выполняется объединение для снижения количества выходных файлов с целью оптимизации записи и последующего чтения (то есть несколько более мелких файлов объединяются в несколько файлов побольше), а также производится дедубликация данных: простой distinct() и затем coalesce(). Результат сохраняется на S3. Эти файлы используются затем для parsing'а , а также являются своеобразным архивом сырых необработанных данных в формате как есть;

   2. Parsing на этой фазе производится разбор входных данных и сохранение их в плоские структуры согласно маппингу, описанному в метаданных. В общем случае из одного входного файла можно получить на выходе несколько плоских структур, которые в виде сжатых (как правило gzip) CSV-файлов сохраняются на S3.

3. Заключительный этап загрузка данных CSV-файлов в ODS-слой хранилища: создается временная external table над данными в S3 через PXF S3 connector, после чего данные уже простым pgsql переливаются в таблицы ODS-слоя Greenplum

4. Все это оркестрируется с помощью Airflow.

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

• создать в ODS-слое Хранилища таблицы-приемники данных;

• в репозитории метаданных в Git согласно принятым стандартам прописать в виде отдельных YAML-файлов:

  • общие настройки источника (такие как: расписание загрузки, входной и выходной формат файлов с данными, S3-бакет, имя сервисного пользователя, имя и email владельца для нотификации в случае сбоя загрузки и т.п.);

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

До недавнего времени такой подход удовлетворял текущие наши потребности, но количество и разнообразие источников данных растет. У нас стали появляться источники, которые не являются реляционными базами данных, а генерируют данные в виде потока JSON-объектов. Кроме того на горизонте уже маячила интеграция источника, который под собой имел MongoDB и поэтому будет использовать MongoDB Kafka source connector для записи данных в Kafka. Поэтому остро встала необходимость доработки нашего frameworkа для поддержки такого сценария. Хотелось, чтобы данные источника сразу попадали на S3 в формате JSON - то есть в формате "как есть", без лишнего шага конвертации в parquet посредством Apache NiFi.

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

df = spark.read.format(in_format) \               .options(**in_options) \               .load(path) \               .distinct()    new_df = df.coalesce(div)new_df.write.mode("overwrite") \             .format(out_format) \            .options(**out_options) \            .save(path)

Но если мы проделаем все те же манипуляции с JSON-данными, то волей-неволей внесем изменения во входные данные, так как при чтении JSONов Spark автоматом определит и сделает mergeSchema, т.е. мы тем самым можем исказить входные данные, чего не хотелось бы. Ведь наша задача на этом шаге только укрупнить файлы и дедублицировать данные, без какого-либо вмешательства в их структуру и наполнение. То есть сохранить их как есть.

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

рассматривать файлы с JSON-объектами как DataFrame с одной колонкой, содержащей весь JSON-объект.

Попробуем сделать это. Допустим, мы имеем следующий файл данных:

file1:

{productId: 1, productName: ProductName 1, tags: [tag 1, tag 2], dimensions: {length: 10, width: 12, height: 12.5}}{productId: 2, price: 10.01, tags: [tag 1, tag 2], dimensions: {length: 10, width: 12, height: 12.5}}

Обратите внимание на формат этого файла. Это файл с JSON-объектами, где 1 строка = 1 объект. Оставаясь, по сути, JSON-ом, он при этом не пройдет синтаксическую JSON-валидацию. Именно в таком виде мы сохраняем JSON-данные на S3 (есть специальная "галочка в процессоре Apache NiFi).

Прочитаем файл предлагаемым способом:

# Читаем данныеdf = spark.read \          .format("csv") \          .option("sep", "\a") \          .load("file1.json")# Схема получившегося DataFramedf.printSchema()root |-- _c0: string (nullable = true)# Сами данныеdf.show()+--------------------+|                 _c0|+--------------------+|{"productId": 1, ...||{"productId": 2, ...|+--------------------+

То есть мы тут читаем JSON как обычный CSV, указывая разделитель, который никогда заведомо не встретится в наших данных. Например, Bell character. В итоге мы получим DataFrame из одного поля, к которому можно будет также применить dicstinct() и затем coalesce(), то есть менять существующий код не потребуется. Нам остается только определить опции в зависимости от формата:

# Для parquetin_format = "parquet"in_options = {}# Для JSONin_format = "csv"in_options = {"sep": "\a"}

Ну и при сохранении этого же DataFrame обратно на S3 в зависимости от формата данных опять применяем разные опции:

df.write.mode("overwrite") \           .format(out_format) \.options(**out_options) \  .save(path)  # для JSON     out_format = "text" out_options = {"compression": "gzip"}  # для parquet   out_format = input_format out_options = {"compression": "snappy"}

Следующей точкой доработки был шаг Parsing. В принципе, ничего сложного, если бы задача при этом упиралась в одну маленькую деталь: JSON -файл, в отличии от parquet, не содержит в себе схему данных. Для разовой загрузки это не является проблемой, так как при чтении JSON-файла Spark умеет сам определять схему, и даже в случае, если файл содержит несколько JSON-объектов с немного отличающимся набором полей, корректно выполнит mergeSchema. Но для регулярного процесса мы не могли уповать на это. Банально может случиться так, что во всех записях какого-то файла с данными может не оказаться некоего поля field_1, так как, например, в источнике оно заполняется не во всех случаях. Тогда в получившемся Spark DataFrame вообще не окажется этого поля, и наш Parsing, построенный на метаданных, просто-напросто упадет с ошибкой из-за того, что не найдет прописанное в маппинге поле.

Проиллюстрирую. Допустим,у нас есть два файла из одного источника со следующим наполнением:

file1 (тот же что и в примере выше):

{productId: 1, productName: ProductName 1, tags: [tag 1, tag 2], dimensions: {length: 10, width: 12, height: 12.5}}{productId: 2, price: 10.01, tags: [tag 1, tag 2], dimensions: {length: 10, width: 12, height: 12.5}}

file2:

{productId: 3, productName: ProductName 3, dimensions: {length: 10, width: 12, height: 12.5, package: [10, 20.5, 30]}}

Теперь прочитаем Sparkом их и посмотрим данные и схемы получившихся DataFrame:

df = spark.read \          .format("json") \          .option("multiline", "false") \          .load(path)df.printSchema()df.show()

Первый файл (схема и данные):

root |-- dimensions: struct (nullable = true) |    |-- height: double (nullable = true) |    |-- length: long (nullable = true) |    |-- width: long (nullable = true) |-- price: double (nullable = true) |-- productId: long (nullable = true) |-- productName: string (nullable = true) |-- tags: array (nullable = true) |    |-- element: string (containsNull = true)+--------------+-----+---------+-------------+--------------+|    dimensions|price|productId|  productName|          tags|+--------------+-----+---------+-------------+--------------+|[12.5, 10, 12]| null|        1|ProductName 1|[tag 1, tag 2]||[12.5, 10, 12]|10.01|        2|         null|[tag 1, tag 2]|+--------------+-----+---------+-------------+--------------+

Второй файл (схема и данные):

root |-- dimensions: struct (nullable = true) |    |-- height: double (nullable = true) |    |-- length: long (nullable = true) |    |-- package: array (nullable = true) |    |    |-- element: double (containsNull = true) |    |-- width: long (nullable = true) |-- productId: long (nullable = true) |-- productName: string (nullable = true)+--------------------+---------+-------------+|          dimensions|productId|  productName|+--------------------+---------+-------------+|[12.5, 10, [10.0,...|        3|ProductName 3|+--------------------+---------+-------------+

Как видно, Spark корректно выстроил схему отдельно для каждого файла. Если в какой-либо записи не было обнаружено поля, имеющегося в другой, то в DataFrame мы видим корректное проставление null (поля price и productName для первого файла).

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

root |-- price: double (nullable = true) |-- productId: long (nullable = true) |-- productName: string (nullable = true)

а во входных данных у нас присутствуют только файлы а-ля file2, где поля price нет ни у одной записи, то Spark упадет с ошибкой, так как не найдет поля price для формирования выходного DataFrame. С parquet-файлами такой проблемы как правило не возникает, так как сам parquet-файл генерируется из AVRO, который уже содержит полную схему данных и, соответственно, эта полная схема есть и в parquet-файле.

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

Таким образом очевидно, что для корректной загрузки данных из JSON-файлов необходимо предопределить схему JSON-файла с данными и читать файлы уже с применением этой схемы. Тогда у нас даже если в JSONе нет какого-то поля, то в самом DataFrame это поле будет, но его значение подставится как null:

df = spark.read \          .format("json") \          .option("multiline","false") \          .schema(df_schema) \          .load(path)

Первая мысль была использовать для хранения схемы имеющийся сервис метаданных - то есть описать схему в YAML-формате и сохранить в имеющемся репозитории. Но с учетом того, что все данные источников у нас проходят через Kafka, решили, что логично для хранения схем использовать имеющийся Kafka Schema Registry, а схему хранить в стандартном для JSON формате (другой формат, кстати говоря, Kafka Schema Registry не позволил бы хранить).

В общем, вырисовывалась следующая реализация:

• Читаем из Kafka Schema Registry схему

• Импортируем ее в pyspark.sql.types.StructType что-то типа такого:

# 1. получаем через Kafka Schema Registry REST API схему данных # 2. записываем ее в переменную schema и далее:df_schema = StructType.fromJson(schema)

• Ну и с помощью полученной схемы читаем JSON-файлы

Звучит хорошо, если бы Давайте посмотрим на формат JSON-схемы, понятной Sparkу. Пусть имеем простой JSON из file2 выше. Посмотреть его схему в формате JSON можно, выполнив:

df.schema.json()  

{    "fields":    [        {            "metadata": {},            "name": "dimensions",            "nullable": true,            "type":            {                "fields":                [                    {"metadata":{},"name":"height","nullable":true,"type":"double"},                    {"metadata":{},"name":"length","nullable":true,"type":"long"},                    {"metadata":{},"name":"width","nullable":true,"type":"long"}                ],                "type": "struct"            }        },        {            "metadata": {},            "name": "price",            "nullable": true,            "type": "double"        },        {            "metadata": {},            "name": "productId",            "nullable": true,            "type": "long"        },        {            "metadata": {},            "name": "productName",            "nullable": true,            "type": "string"        },        {            "metadata": {},            "name": "tags",            "nullable": true,            "type":            {                "containsNull": true,                "elementType": "string",                "type": "array"            }        }    ],    "type": "struct"}

Как видно, это совсем не стандартный формат JSON-схемы.

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

как сохранить схему уже прочитанного DataFrame в JSON, затем использовать повторно

либо на репозиторий https://github.com/zalando-incubator/spark-json-schema, который нам бы подошел, если мы использовали Scala, а не pySpark

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

К счастью, у нас уже был один источник, генерирующий данные в формате JSON. Как временное решение схема его интеграции в DataPlatform была незамысловата: NiFi читал данные из Kafka, преобразовывал их в parquet, использую прибитую гвоздями в NiFi схему в формате AVRO-schema, и складывал на S3. Схема данных была действительно непростой и с кучей вложенных структур и нескольких десятков полей - неплохой тест-кейс в общем-то:

root |-- taskId: string (nullable = true) |-- extOrderId: string (nullable = true) |-- taskStatus: string (nullable = true) |-- taskControlStatus: string (nullable = true) |-- documentVersion: long (nullable = true) |-- buId: long (nullable = true) |-- storeId: long (nullable = true) |-- priority: string (nullable = true) |-- created: struct (nullable = true) |    |-- createdBy: string (nullable = true) |    |-- created: string (nullable = true) |-- lastUpdateInformation: struct (nullable = true) |    |-- updatedBy: string (nullable = true) |    |-- updated: string (nullable = true) |-- customerId: string (nullable = true) |-- employeeId: string (nullable = true) |-- pointOfGiveAway: struct (nullable = true) |    |-- selected: string (nullable = true) |    |-- available: array (nullable = true) |    |    |-- element: string (containsNull = true) |-- dateOfGiveAway: string (nullable = true) |-- dateOfGiveAwayEnd: string (nullable = true) |-- pickingDeadline: string (nullable = true) |-- storageLocation: string (nullable = true) |-- currentStorageLocations: array (nullable = true) |    |-- element: string (containsNull = true) |-- customerType: string (nullable = true) |-- comment: string (nullable = true) |-- totalAmount: double (nullable = true) |-- currency: string (nullable = true) |-- stockDecrease: boolean (nullable = true) |-- offline: boolean (nullable = true) |-- trackId: string (nullable = true) |-- transportationType: string (nullable = true) |-- stockRebook: boolean (nullable = true) |-- notificationStatus: string (nullable = true) |-- lines: array (nullable = true) |    |-- element: struct (containsNull = true) |    |    |-- lineId: string (nullable = true) |    |    |-- extOrderLineId: string (nullable = true) |    |    |-- productId: string (nullable = true) |    |    |-- lineStatus: string (nullable = true) |    |    |-- lineControlStatus: string (nullable = true) |    |    |-- orderedQuantity: double (nullable = true) |    |    |-- confirmedQuantity: double (nullable = true) |    |    |-- assignedQuantity: double (nullable = true) |    |    |-- pickedQuantity: double (nullable = true) |    |    |-- controlledQuantity: double (nullable = true) |    |    |-- allowedForGiveAwayQuantity: double (nullable = true) |    |    |-- givenAwayQuantity: double (nullable = true) |    |    |-- returnedQuantity: double (nullable = true) |    |    |-- sellingScheme: string (nullable = true) |    |    |-- stockSource: string (nullable = true) |    |    |-- productPrice: double (nullable = true) |    |    |-- lineAmount: double (nullable = true) |    |    |-- currency: string (nullable = true) |    |    |-- markingFlag: string (nullable = true) |    |    |-- operations: array (nullable = true) |    |    |    |-- element: struct (containsNull = true) |    |    |    |    |-- operationId: string (nullable = true) |    |    |    |    |-- type: string (nullable = true) |    |    |    |    |-- reason: string (nullable = true) |    |    |    |    |-- quantity: double (nullable = true) |    |    |    |    |-- dmCodes: array (nullable = true) |    |    |    |    |    |-- element: string (containsNull = true) |    |    |    |    |-- timeStamp: string (nullable = true) |    |    |    |    |-- updatedBy: string (nullable = true) |    |    |-- source: array (nullable = true) |    |    |    |-- element: struct (containsNull = true) |    |    |    |    |-- type: string (nullable = true) |    |    |    |    |-- items: array (nullable = true) |    |    |    |    |    |-- element: struct (containsNull = true) |    |    |    |    |    |    |-- assignedQuantity: double (nullable = true) |-- linkedObjects: array (nullable = true) |    |-- element: struct (containsNull = true) |    |    |-- objectType: string (nullable = true) |    |    |-- objectId: string (nullable = true) |    |    |-- objectStatus: string (nullable = true) |    |    |-- objectLines: array (nullable = true) |    |    |    |-- element: struct (containsNull = true) |    |    |    |    |-- objectLineId: string (nullable = true) |    |    |    |    |-- taskLineId: string (nullable = true)

Естественно, я не захотел перебивать руками захардкоженную схему, а воспользовался одним из многочисленных онлайн-конвертеров, позволяющих из Avro-схемы сделать JSON-схему. И тут меня ждал неприятный сюрприз: все перепробованные мною конвертеры на выходе использовали гораздо больше синтаксических конструкций, чем понимала первая версия конвертера. Дополнительно пришло осознание, что также как и я, наши пользователи (а для нас пользователями в данном контексте являются владельцы источников данных) с большой вероятностью могут использовать подобные конвертеры для того, чтобы получить JSON-схему, которую надо зарегистрировать в Kafka Schema Registry, из того, что у них есть.

В результате наш SparkJsonSchemaConverter был доработан появилась поддержка более сложных конструкций, таких как definitions, refs (только внутренние) и oneOf. Сам же парсер был оформлен уже в отдельный класс, который сразу собирал на основании JSON-схемы объект pyspark.sql.types.StructType

У нас почти сразу же родилась мысль, что хорошо бы было поделиться им с сообществом, так как мы в Леруа Мерлен сами активно используем продукты Open Source, созданные и поддерживаемые энтузиастами со всего мира, и хотим не просто их использовать, но и контрибьютить обратно, участвуя в разработке Open Source продуктов и поддерживая сообщество. В настоящий момент мы решаем внутренние орг.вопросы по схеме выкладывания данного конвертера в Open Source и, уверен, что в ближайшее время поделимся с сообществом этой наработкой!

В итоге благодаря написанному SparkJsonSchemaConverterу доработка шага Parsing свелась только к небольшому тюнингу чтения данных с S3: в зависимости от формата входных данных источника (получаем из сервиса метаданных) читаем файлы с S3 немного по-разному:

# Для JSONdf = spark.read.format(in_format)\            .option("multiline", "false")\            .schema(json_schema) \            .load(path)# Для parquet:df = spark.read.format(in_format)\            .load(path)

А дальше отрабатывает уже существующий код, раскрывающий все вложенные структуры согласно маппингу и сохраняющий данные DataFrameа в несколько плоских CSV-файлов.

В итоге мы смогли при относительном минимуме внесенных изменений в код текущего frameworkа добавить в него поддержку интеграции в нашу Data Platform JSON-источников данных. И результат нашей работы уже заметен:

• Всего через месяц после внедрения доработки у нас на ПРОДе проинтегрировано 4 новых JSON-источника!

• Все текущие интеграции даже не заметили расширения функционала frameworkа, а значит, мы точно ничего не сломали произведенной доработкой.