Чем покупка букета на 8 Марта через Яндекс Еду отличается от покупки, собственно, еды? С точки зрения пользователя — ничем. Выбрал, оплатил, доставили. А вот с точки зрения разработчика бэкенда заказ уникальных букетов превращается в нетривиальную инженерную задачу синхронизации складских запасов. Задержка синхронизации хотя бы в 10 минут трансформируется в звонок и сборщиков заказов, сообщающих о том, что именно такого букета на складе больше нет.
Меня зовут Виталий Московкин, я занимаюсь ритейлом в Яндекс Еде. В статье я расскажу, как мы синхронизировали состояние складов с 18 миллионами уникальных товаров: сначала с помощью PostgreSQL, а затем с помощью YDB. Такое количество товаров превращается на бэкенде в 4 миллиарда записей о ценах и стоках, которые нельзя просто так кешировать. Но и замена монолитной СУБД на распределённую тоже задача не на десять минут. Подробности — ниже.
Глава первая, в которой в Яндекс Еде появляются букеты
Ритейл в Яндекс Еде — это отдельный проект. При заказе еды из ресторана эту еду приготовят после размещения заказа. А при заказе продуктов с полок в супермаркетах сервису нужно заранее знать, что находится на этих полках.
При старте проекта сервис располагал данными о 6 миллионах уникальных товаров в 500 000 категорий. Так как один и тот же товар мог быть в разных магазинах, то суммарно получалось порядка 1 миллиарда записей о ценах и остатках.
Первая версия сервиса была собрана на базе PostgreSQL и представляла собой гигантский кеш. Который, с одной стороны, обновлялся выгрузкой данных от партнёров, а с другой стороны, читался остальными сервисами, например витриной, поиском или аналитикой.
JSON'ы по 100 мегабайт
Как это обычно бывает в ритейле, магазины позволяли получать по API текущее состояние склада в виде JSON. Сервис периодически опрашивал всех партнёров, парсил полученные JSON'ы и заменял соответствующие сегменты в базе данных с помощью DELETE и INSERT.
Такая схема хорошо работала, пока магазинов было немного. Но каждый месяц их становилось всё больше, а в какой‑то момент продуктовая команда ещё и выдвинула ультиматум: «Цветочников на 8 Марта парсим раз в 5 минут». При такой нагрузке база начала захлёбываться от write amplification, постоянно перезаписывая гигабайты данных ради обновления нескольких ценников.
Команда поняла, что простую архитектуру MVP пора менять на что‑то более сложное. Бизнес хотел не только актуального состояния букетов, но и другие фичи: фильтрацию данных, обогащение изображений товаров и описаний, ML‑аналитику. Добавление всего этого в парсер JSON привело бы к бесконтрольному накоплению технического долга.
Глава вторая, в которой «слоны» обновлений разделяются на части
Пересмотрев доклады с HighLoad++ и учтя накопившийся опыт, команда выбрала для сервиса новую архитектуру, в которой «чистые» данные были отделены от витрин и продуктовой логики.
Выделенное хранилище
Основой архитектуры стало «простое» хранилище, умеющее надёжно хранить байты и считать дельты, но ничего не знающее о смысле хранимых данных. Благодаря такому хранилищу нет соблазна добавлять внутрь базы сложные фильтры и настройки продукта, что обычно приводит к создания «чёрных ящиков», в которых товар может исчезнуть с витрины, и никто, от поддержки до CTO, не сможет быстро ответить почему.
Работу с продуктовыми правилами и логикой мы вынесли в отдельные сервисы. Один из таких сервисов — предыдущая реализация на PostgreSQL, которая перестала быть источником правды о данных и стала просто одним из потребителей хранилища. Сервис с кешем категорий позволил скрывать пустые разделы меню. Сервис рекомендаций использовал собственный легковесный индекс, чтобы предлагать только то, что есть в наличии. А сервисы‑агрегаторы собирали «мастер‑карточки» с минимальной ценой по региону, например для «„Нурофена“ в Москве». Всё это — поверх одного хранилища и с использованием одной кодовой базы.
Выделив хранилище, мы поменяли то, как данные попадают в базу. Отдельный сервис получает данные от партнёров, разбивает JSON'ы и помещает записи в топик брокера сообщений. Хранилище вычитывает поток записей и обновляет информацию о товарах.
Первая версия хранилища могла обновлять запись о товаре двумя способами. Вычитывая из стриминговой платформы команду PATCH, можно было обновить часть полей, а с помощью команды SAVE запись обновлялась целиком. Но довольно быстро мы поняли, что ни один из сервисов не знает информацию о записи целиком. Сервис интеграции цен знает только цены, сервис интеграции стоков — только количество товара на складе, и оба они ничего не знают о, например, изображении товара.
Хранилище написано на Go, и код работы с потоком записей получился компактным, во многом благодаря дженерикам. Сервис «склеивает» пазл, не зная структуры данных TRecord:
func (s *StorageServiceImpl[...]) patchRecord(record TRecord, patch TRecord) TRecord { // Storage не знает бизнес-логики. // Он просто берёт старые байты и накатывает на них новые байты. patchedState := safeMerge(record.GetState(), patch.GetState()) record.SetState(patchedState) patchedData := safeMerge(record.GetData(), patch.GetData()) record.SetData(patchedData) // ...мёржим поколения и флаги удаления... return record }
Вычитав запись из брокера сообщений, хранилище проверяет, изменилось ли итоговое состояние товара. Если это так, то в исходящий топик CHANGES записывается информация об изменении: полная информация о товаре до и после изменения. Другим сервисам достаточно подписаться на этот топик: они могут быть stateless и не читать данные напрямую из хранилища, что существенно снижает нагрузку на систему.
Для обеспечения надёжности система использует принцип keyframe, похожий на то, как это реализовано в видеокодеках: раз в сутки полный слепок всех данных выгружается в отдельный топик. Это позволяет всем потребителям «сверять часы» и устранять возможный дрифт данных. При этом основной топик не «забивается» и продолжает использоваться для получения изменений в реальном времени.
Благодаря использованию брокера сообщений и хранилища система создаёт сетевую нагрузку, кратную изменениям данных. Если из 10 тысяч товаров в магазине цена изменилась только у 100, то в топик CHANGES будет записана информация только об этих товарах. По сравнению с первой реализацией, сетевой трафик падает в сотни раз.Дополнительным бонусом стало то, что когда партнёры начали отдавать честные дельты товарных остатков, архитектура была к этому полностью готова.
Новая СУБД
Разрабатывая вторую версию системы, мы ответственно подошли к выбору СУБД и перебрали всё меню Яндекса: самописную автоматику поверх Managed Postgres, MySQL, динтаблицы в YTsaurus, ClickHouse. Выбрали YDB (СУБД Яндекса):
Автоматическое решардирование. YDB разбивает обработку данных каждой таблицы на обработчики‑таблетки. Каждый из обработчиков отвечает за свою часть таблицы: если нагрузка возрастает, то количество обработчиков увеличивается, а если снижается — то количество уменьшается.
Автоматическое масштабирование. YDB разделяет слои хранения и вычисления. Добавление новых серверов в слой вычисления автоматически запускает на них выполнение таблеток, линейно масштабируя производительность как на чтение, так и на запись.
Брокер сообщений с exactly once. YDB можно использовать как PostgreSQL для строковых таблиц, можно как ClickHouse для колоночных, а можно как Kafka для топиков. В режиме брокера сообщений YDB может перекладывать данные между таблицами и топиками с гарантиями exactly once. В будущем эта возможность позволит нам упростить архитектуру и перестать использовать сложные архитектурные решения вроде Transactional Outbox.
Глава третья, в которой JSON превращается в protobuf
Масштабируемая СУБД и передача дельт с помощью брокера сообщений — надёжный фундамент новой системы. Но я бы не сел писать статью на «Хабр», если бы всё было так просто.
Почему protobuf, а не широкие таблицы?
Архитектура «простое хранилище» подразумевает, что хранилище ничего не знает о хранимых данных. Тем не менее эти данные нужно хранить и применять к ним изменения. В первой версии сервиса мы использовали JSON, и у него есть проблема с избыточностью. Бесконечные повторения {"price": 100, "stock": 5} создают нагрузку на систему хранения, даже если воспользоваться не текстовым, а бинарным форматом хранения JSON.
Можно создать в таблице много колонок под разные поля описания товара. Но в ритейле схема данных меняется слишком часто: сегодня добавили discount_price, завтра — is_eco_friendly. А постоянные миграции терабайтных таблиц на новую схему потребуют усложнения архитектуры, не говоря уже о дополнительной нагрузке на СУБД.
Поэтому в основной таблице нашего хранилища всего девять колонок, из которых главную роль играют колонка первичного ключа и колонка data с сериализованным Protobuf:
CREATE TABLE place_product ( -- Первичный ключ (про hash будет отдельная драма ниже) hash Int64 NOT NULL, brand_id Int64 NOT NULL, place_id Int64 NOT NULL, product_id Utf8 NOT NULL, -- Пэйлоад (Protobuf Blob) data String NOT NULL, -- Метаданные (вынесли только то, что нужно для индексов) state String NOT NULL, -- Статус («Доступен», «Проверен», «Скрыт») generation Int64, -- Для механизма Generations created_at Int64 NOT NULL, updated_at Int64 NOT NULL, PRIMARY KEY (hash, brand_id, place_id, product_id) ) WITH ( AUTO_PARTITIONING_BY_SIZE = ENABLED, AUTO_PARTITIONING_BY_LOAD = ENABLED );
Protobuf — максимально компактный бинарный формат в котором, в отличие от JSON'а, не хранятся текстовые имена полей. Когда в описание товара добавляется новое поле, то мы обновляем единый proto‑файл и пересобираем все сервисы.
Описывая структуры данных в proto‑файлах, мы сохраняем обратную совместимость: поставщик данных может обновиться раньше потребителя, и сервис не должен сломаться, получив новую версию данных. Protobuf позволяет так делать, если не переиспользовать номера полей и радикально не менять типы.
Сервис хранилища обновляется вместе с поставщиками данных: это нужно для того, чтобы механизм слияния proto.Merge смог работать с новыми полями в Go‑структурах. Это критично, например, для списков: мы хотим заменять их целиком, а не дописывать новые значения в хвост к старым — что произошло бы с неизвестными полями.
Как мы удаляем неактуальные товары
Переход от архитектуры «перезаписываем всё» к архитектуре «применяем изменения» выявил неожиданную проблему. Если партнёр перестал присылать информацию о товаре, то как хранилище узнает, что этого товара больше нет на складе?
Чтобы решить эту проблему, мы воспользовались механизмом поколений (Generations), который работает по принципу Mark‑and‑Sweep из сборщиков мусора. Добавили к каждой записи поле generation -- timestamp начала импорта. Алгоритм работы получился такой:
Начинаем импорт от партнёра. Фиксируем время старта
T_current.Все записи, которые приходят в этом сеансе, мы сохраняем (
Upsert/Patch) сgeneration = T_current.Когда импорт закончен, запускаем фоновую «чистку»
(PATCH_GENERATION). Мы говорим базе: «Найди все товары этого магазина, у которыхgeneration < T_current, и проставь им статусUnavailable».
Работая с потоком изменений, мы сохраняем гарантии полной консистентности снапшота. То, что пропало из источника, исчезнет и из базы.
Синхронизация асинхронных обновлений и валидаций
Используя брокер сообщений в высоконагруженной распределённой системе, мы не можем рассчитывать на строго определённый порядок обработки сообщений. Типичный «проблемный» сценарий выглядит вот так:
Интеграция шлёт: «
Цена = 100» (версия 1).Валидатор видит «100», понимает, что это ошибка, и шлёт: «
Скрыть товар» (базируясь на версии 1).В этот момент интеграция шлёт исправленную цену: «
Цена = 200» (версия 2).
Если сообщение от валидатора (2) придёт позже исправления (3), мы скроем хороший товар с ценой 200, потому что валидатор думал, что там всё ещё 100.
Для решения этой и подобных проблем мы применяем ту же стратегию, которую внутри использует сама YDB: оптимистические блокировки. А для версионирования используем поле updated_at.
Интеграции отправляют данные с updated_at в значении nil, всегда перезаписывая поля в базе. А валидаторы отправляют изменение с условием: «Примени это, только если текущий updated_at в базе равен тому, который я видел, когда принимал решение» (такая реализация оптимистических блокировок обычно называется Compare‑and‑Swap, CAS).
Если за время работы валидатора запись успела измениться (пришла новая цена), условие не сработает, и патч не будет применён. Товар не скроется ошибочно, а валидатор просто пересчитает статус на следующем прогоне, когда увидит актуальную цену.
Батчевание сообщений на уровне protobuf
Брокеры сообщений традиционно работают с батчами (группами) сообщений гораздо быстрее, чем с записью и чтением по одному. Мы не стали полагаться на настройки конкретного брокера и сделали группировку явным констрактом системы. В коде StorageService мы накапливаем сразу 1000 обновлений и записываем в топик один protobuf‑объект типа Batch<Record>:
// storage_service.go: jobPublish func (s *StorageServiceImpl[...]) jobPublish(...) error { // ...собираем messages из job.tasks... // Формируем ОДИН батч из множества сообщений batchMessage := s.createBatchOutMessage(messages) // И отправляем его как единый атом return s.diffPublisher.Publish(ctx, batchMessage) }
При таком подходе накладные расходы падают в 1000 раз. Брокер записывает всего одно сообщение: один заголовок, один ack, один fsync со стороны брокера. А потребитель получает сразу много сообщений и итерируется по ним с помощью for item in batch.Items. В stateless‑архитектуре обрабатывать такие сообщения можно параллельно.
Глава четвёртая, в которой мы учимся работать с распределённой СУБД
Сервис хранилища мы написали на Go за три недели, примерно 10 тысяч строк кода. С чем мы столкнулись во время отладки?
Cross‑Shard Transactions
Первые же тесты на небольшом бренде цветочников показали неожиданно высокую нагрузку на СУБД. Оказалось, что запрос UPDATE, передающий много ключей в INNER JOIN, — это не то, как принято работать с распределёнными системами. Планировщик YDB не знал, на каких шардах находятся данные, поэтому каждый запрос выполнялся на десятках узлов СУБД, опрашивая весь кластер ради добавления 1000 строк.
Вот так выглядел JOIN в первой версии запроса:
select ... from AS_TABLE($parameters) as p inner join place_product as e on p.brand_id = e.brand_id AND ...
Мы переписали запрос, в явном виде передавая оптимизатору полный список первичных ключей. Получив такой список, планировщик YDB выполняет запрос только на нужных узлах, что превращает тяжёлые распределённые транзакции в серию лёгких локальных чтений:
$keys = ListMap($parameters, ($p) -> { return (p.brand_id, p.place_id, p.product_id) }); $existing = ( select * from place_product where (brand_id, place_id, product_id) in $keys ); -- И уже потом работаем с $existing в памяти запроса
Более того, мы научили Storage перегруппировывать записи перед записью. Вместо того, чтобы кидать в базу разношёрстный батч, который разлетится по всему кластеру, мы стараемся собрать записи так, чтобы они попадали в один шард (или в соседние). Это превращает тяжёлую распределённую запись в серию быстрых параллельных локальных коммитов.
Исправив код, мы запустили фоновую заливку. Сначала встроенный мониторинг YDB показывал стабильную скорость записи, но через некоторое время она неожиданно увеличилась! Оказалось, что по мере увеличения размера таблиц YDB начала автоматически «делить» обслуживающие их таблетки (size based splitting). В результате получилось, что чем больше данных мы заливали в базу, тем лучше она работала.
Автоматическое решардирование по Primary Key
После успеха первых экспериментов мы решили залить данные одной из сетей супермаркетов (это тысячи магазинов). Вначале всё шло хорошо, но затем несколько узлов кластера ушли в 100%‑е потребление процессора, а остальные при этом простаивали.
Проблема оказалась в первичном ключе. Изначально он начинался с BrandID: Primary Key = (BrandID, PlaceID, ProductID). Данные в YDB упорядочены лексикографически по первичному ключу. Это значит, что все товары одного бренда лежат физически рядом, часто на одном шарде.
Когда мы начали лить обновления для супермаркетов, мы создали то, что называется «горячей партицией». Так как первичный ключ увеличивался линейно, то автоматическое решардирование никак не помогало снижать нагрузку: таблетки исправно «делились», но новые записи всегда попадали на одну и ту же таблетку.
Коллеги из соседней команды подсказали, что нужно равномерно «размазать» записи по партициям‑таблеткам с помощью добавления «соли» к ключу шардирования. Мы изменили Primary Key, поставив в начало хеш от ID магазина: Primary Key = (Hash(BrandID, PlaceID), BrandID, PlaceID, ProductID).
Хеш равномерно распределил записи по шардам, нагрузка на запись выровнялась, и мы снова получили тот самый линейный рост, ради которого и выбирали YDB.
Generation и оптимистические блокировки
Когда мы добавили в архитектуру механизм поколений, то столкнулись с логической ошибкой, которую не отловили на тестах: generation обновляется часто — при каждом парсинге, раз в 5–10 минут; а реальные данные цены/стока — редко.
Каждый парсинг порождал события в исходящем топике, и мы DDoS'или сами себя миллиардами пустых событий «поколение изменилось». Но если не слать такие события и при этом обновлять updated_at в базе, то оптимистические блокировки начинают влиять на внешние сервисы.
Например, валидатор пытается скрыть товар, используя updated_at, который он видел последним (например, 12:00). Но в базе уже лежит updated_at = 12:05 (потому что в 12:05 прошёл парсинг и обновил поколение). Валидатор получает отказ (Conflict), хотя данные товара не менялись!
Получается ситуация, когда валидатор не может обновить запись, потому что фоновые обновления поколений иногда успевают обогнать его. Это происходит не всегда, но достаточно часто. А отлаживать такие плавающие и трудновоспроизводимые проблемы в логике долго и сложно.
Мы поменяли логику так, чтобы запись в исходящий топик добавлялась только в случае изменения цены или остатка товара. При изменении поколения мы меняем поле Generation в базе, но не трогаем поле updated_at и не добавляем запись в исходящий топик.
Так мы получили систему, где технические апдейты (поколения) не мешают продуктовым (цена/наличие).
Цифры с прода
Вот так выглядят наши таблицы сейчас. Обратите внимание на цифры: это прод.
База живёт, дышит и сама себя балансирует.
Глава пятая, в которой мы подводим итоги
Мы переписали core‑функциональность, поменяли СУБД и архитектуру системы. Стоило ли оно того?
Сжатие трафика в 100 раз
Лучше всего изменения видны на сетевой активности системы.
Мы отфильтровали 99% информационного шума. Потребители данных теперь получают только реальные изменения цен и стоков. Им не нужно обрабатывать гигабайты одинаковых JSON'ов, чтобы найти одно изменение.
Единый источник данных для быстрого решения продуктовых задач
Цены на цветы 8 Марта обновляются раз в 5 минут, а телефон поддержки молчит.
Новые микросервисы (например, «Подборки товаров со скидками») теперь подключаются к одному готовому топику изменений. Нам не нужно строить для них отдельные механизмы выгрузки или специальные ручки API. Есть единый поток данных, на который можно подписаться и вычитывать обновления. Подключение нового сервиса — это часы, а не недели разработки.
YDB — это не drop‑in‑замена для PostgreSQL или MariaDB
Да, распределённую природу этой СУБД нужно учитывать. Но в остальном YDB показала себя отлично. Она линейно масштабируется: когда нам не хватало мощности, мы просто доливали железо, и кластер сразу же начинал его использовать без даунтайма и ручного решардинга.
Поток обновлений для удобного подключения новых сервисов
Самое главное — мы получили гибкую машину состояний.
Хотим добавить ML‑прогноз спроса? Нам не нужно лезть в код интеграции или переписывать ядро. Мы просто ставим рядом отдельный микросервис, который слушает поток обновлений, считает прогноз и шлёт обратно в хранилище маленький PATCH с новыми полями. Хранилище само склеивает полученный PATCH с ценой от партнёра и статусом от валидатора.
Мы построили конструктор, к которому можно подключать новые модули бизнес‑логики буквально за день.
Актуальность данных — это лояльность пользователей. Когда вы говорите клиенту правду — «товара нет», он может расстроиться, но останется с вами. Когда вы обманываете его и говорите, что товар есть, а потом звоните с отменой — он уходит навсегда.
Не бойтесь менять архитектуру, если старая не справляется. Иногда, чтобы спасти цветы на 8 Марта, нужно разобрать старый, уютный, но безнадёжно устаревший заводик и построить современный комбинат по переработке данных.
А как вы боретесь с Write Amplification и обеспечиваете консистентность в своих высоконагруженных системах? Используете Redis, Tarantool, ClickHouse или всё ещё верите в мастер‑мастер‑репликацию MySQL? Жду вас в комментариях — обсудим ваши истории!
