Как связаны скидки, пользовательские пути и огромные массивы данных в Яндекс Рекламе?
Привет, Хабр! Меня зовут Максим Стаценко, я работаю с базами данных и яростно в них копаюсь с 2010 года, а в Big Data — с 2016. Сейчас работаю в Яндексе в DWH поиска и рекламы.
Мы работаем с ОЧЕНЬ большими данными. Каждый день миллионы пользователей видят рекламу Яндекса, а наши системы обрабатывают огромные объёмы данных. Чтобы реклама работала эффективно, нам нужно в каждый момент времени иметь максимально полную информацию об истории жизни рекламного объявления, а значит нужно каким-то образом передавать данные от одного события к другому внутри рекламной воронки. Расскажу, как мы решали эту проблему.
Проблема воронки
Почти у любого сервиса — своя воронка. Например, вы показываете скидку на лендинге, но пользователь оформляет покупку не сразу, а побродив по сайту нажимает на заветную кнопку «купить» на другом экране. Как передать информацию о скидке на лендинге, так чтобы она не потерялась к моменту нажатия на кнопку? И это только один пример — в Яндексе масштабы намного больше, и задачи ещё сложнее.
Каждый сервис ищет свои решения. Одни выбирают классический путь: пишут множество SELECT’ов, соединяют таблицы через JOIN’ы и собирают данные вручную. Например, берут таблицу заходов на сайт, склеивают её с таблицами добавления в корзину и оплаты — и вот вам воронка. Но такой подход — это боль: запросы громоздкие, выполнение долгое, а масштабируемость страдает. Нам нужно что-то быстрее, проще и мощнее.
Другой способ — передавать данные через GET- или POST-параметры. То есть, при переходе на следующий этап воронки нужная информация просто прикрепляется к URL или уходит в теле запроса. Звучит удобно, но на деле этот метод тоже имеет кучу проблем.
Мы обеспечиваем это через Stateful Steaming в бэкенде. Это работает быстро и не слишком нагружает аналитику и бэкэнд.
Яндекс Реклама
Реклама — это тоже воронка.
Когда вы заходите на сайт, он отправляет запрос на сервер Яндекса: «Эй, Яндекс, какую рекламу показать этому пользователю?» Яндекс запускает мгновенный аукцион среди всех доступных объявлений, выбирает самые релевантные, а затем среди них — то, которое принесёт наибольший доход. Получив ответ, сайт оценивает, а стоит ли игра свеч. Если условия устраивают, реклама появляется на экране.
Как только ваш браузер отобразил рекламу, он отправляет сигнал в Яндекс, и система фиксирует показ. Если реклама пользователю действительно полезна — он на неё кликнет. Но перед тем как попасть на целевой сайт, клик проходит через редирект через серверы Яндекса. Так платформа узнаёт, что объявление сработало.
Так на стороне Яндекса есть воронка из трёх этапов:
Хит, когда сайт запросил рекламу у Яндекса.
Показ рекламы.
Клик по рекламе.
На самом деле этапов больше, и где-то среди них есть тот, на котором сосредоточен наш интерес — антифрод.
Мы не хотим тратить деньги пользователей на просмотр рекламы роботами, это несправедливо, поэтому у нас есть антифрод. Один — быстрый, обрабатывает за 30 минут, другой — медленный, работает 100 дней. То есть если в течение 100 дней после клика Яндекс решил, что это всё-таки кликнул робот, то мы вернём деньги и откатим все изменения.
Устройство и проблемы старой системы
Раньше система была устроена достаточно примитивно.
Данные передавались по цепочке: из хита — в показ, из показа — в клик, а дальше, например, в антифрод. Но возникала проблема: если внезапно понадобилось жирное поле referer, чтобы отследить, откуда пришёл пользователь (вдруг он с подозрительного сайта, который стоит забанить), то его нужно было протащить через весь pipeline. И это уже было не просто.
Сначала нужно было забросить referer из хита в показ, потом эту жирную строку перебросить в клик, и дальше из клика передать в антифрод.
Первая очевидная проблема — проблема сломанного телефона.
Вначале отправляем referer, а дальше идёт преобразование кодом, и так несколько раз. На выходе получаем не пойми что.
Вторая проблема — сильно загруженный runtime. Если вы возьмёте Яндекс Рекламу, нажмёте правой кнопкой и скопируете ссылку, увидите примерно такое полотно:
Всё, что находится после /count — это закодированные передающиеся данные. Если вы вместили сюда referer, то во все рекламные URL’ы на сайте добавится ещё на закодированный referer. Это замедляет и сайты, и работу Яндекса — неудобно для пользователей. Поэтому бэкендеры запретили прокидывать большие поля.
Решение: большие поля прокидываем только через JOIN.
Сложность в том, что воронка растягивается на 100 дней. Это значит, что нам приходится каждый день собирать и соединять огромные таблицы (хиты, показы, клики) за разные даты. А теперь представьте масштаб: каждая такая таблица весит чуть больше терабайта. А JOIN этих гигантов — настоящий монстр, который съедает тонны CPU и тормозит всю систему. Всё работает медленно, анализ данных запаздывает, а мы теряем драгоценное время на реакцию.
Но есть и третья проблема. В Яндекс приходят умные люди, и каждый придумывает свой уникальный способ соединить эти три лога. В результате ответы получаются самые разные.
Пример из нашей жизни:
Представьте: CTO приходит в разгар разбирательства с советом директоров и просит подсчитать количество реклам мобильных приложений. Все бросают всё и бегут считать. Через полчаса CTO получает три разных ответа, причём отличающихся на порядок. Очевидно, что эту проблему нужно решать. И она — не единственная.
Проблемы старой схемы:
Нагружает runtime.
Большие данные не прокинешь – нужны JOIN’ы в MapReduce.
Жжём железо в MapReduce.
Чтобы прокинуть поле, нужен разработчик. Либо DWH-специалист, который напишет эти JOIN’ы, либо бэкендер, который напишет в передачу pipeline, кодирование в URL, и т.д. Если хотите поместить в ML новую фичу, то идите в backlog и ждите месяц.
Много железа.
-
Трудно дебажить:
- Иногда до конца цепочки доходит не то, что было в начале.
- Иногда JOIN пишут по-разному и получают разные ответы на один вопрос.
С этим всем надо было что-то делать.
Как оптимизировали работу с данными в новой системе
Мы с командой придумали новое решение для хранения состояния (State).
Теперь есть Key Value Storage, в нашем случае это YTSaurus DynTables, но по сути он похож на HBase или Cassandra. Мы решили хранить в нём всю историю одного запроса за рекламой.
Как это работает:
Сайт запрашивает рекламу → мы генерируем хит и создаём для него уникальный ID.В Key-Value Storage сохраняем все события по этому хиту — каждое в своей колонке в виде массива:
Если было два показа → в колонке хранится массив из двух записей.
Если было три клика → хранятся три события клика в массиве.
Таким образом, вся цепочка действий хранится в одном месте и доступна для анализа.
Теперь разберёмся, как именно мы записываем данные в хранилище. Чтение входа выглядит так:
Здесь никакого rocket science. Все события сначала попадают в Kafka — это наша шина, Решардеры и сервисы читают эти события и записывают в Key Value Storage. Если по хиту уже есть данные, мы не перезаписываем, а обновляем массив событий. Так мы гарантируем целостность данных, сохраняя всю историю запроса в одном месте.
Интереснее, как передавать данные наружу. Запись выхода выглядит так:
Не все сервисы могут напрямую ходить в Key-Value Storage, поэтому нужно как-то раздавать данные тем, кто привык их получать. Наши менеджеры стейта пишут в очередь notify. Как только по конкретному хиту происходит событие, например, показ или вердикт антифрода, отправляется уведомление. Сервисы-писатели подписаны на очередь и читают только те события, которые нужны сервису: антифроду важны клики и показы — сами хиты ему неинтересны. Другим компонентам нужны свои данные. Так мы избавляемся от избыточных запросов, и каждый получает только то, что ему нужно.
Когда приходит событие о показе, подписанный сервис-писатель идёт в Key-Value Storage и делает lookup по ID. Там выцепляет нужные данные, например, историю хита, реферер, параметры кампании. Затем отправляет их в нужный сервис, который уже знает, что делать. Так каждый получает только актуальную информацию без лишней нагрузки на хранилище.
Схема классная: быстро работает, на наших технологиях быстро собирается. Но есть одна проблема — ресурсы. Один день данных в этом Key Value Storage занимает 1,25 PB, 100 дней — 25 PB.
Я пошёл защищать схему к CTO, но мне сказали, что она слишком дорогая. Хотя идея классная, не хотелось её отвергать. Поэтому начали думать дальше. Наше внимание привлёк график частоты обновлений старых событий. Очевидно, что они обновляются логарифмической кривой:
Чем свежее событие, тем чаще оно обновляется. А чем старше, тем реже в нём что-то меняется. Ведь редко бывает, что пользователь загрузил сайт и только через 50 дней кликнул на рекламу — такие сценарии единичны. Тогда мы задумались: а нужно ли вообще отправлять старые события так же быстро, как новые?
Оказывается, антифроду плевать, если событие произошло 30 дней назад: получит ли он его через минуту или пять. А значит, мы можем снижать время отклика на старых событиях. В Яндексе для этого есть несколько видов кластеров.
OLTP-кластер, на котором работает весь production. Находится в трёх дата-центрах, очень быстрый. В нём стоят быстрые сервера и диски, но он самый дорогой по ресурсам.
Аналитические кластеры, рассчитанные под OLAP-нагрузку от ML и аналитиков — людей, которые часто пишут неоптимальные большие запросы, а затем ждут сутками, пока они посчитаются. Аналитические кластеры хороши тем, что, в них много места, они заточены под MapReduce, и множество HDD (жёстких дисков). Но и у них есть проблема — они находятся в одном дата-центре. Поэтому таких кластеров нам нужно два — если с одним дата-центром что-то произойдёт, нужен резервный.
Мы решили разнести нагрузку на два типа кластеров. Свежие события за первые 48 часов храним на быстром OLTP-кластере. Он работает на SSD и мгновенно отдаёт данные. Старые события переносим в медленный OLAP-кластер на HDD. Так мы обеспечиваем скорость для актуальных данных, экономим ресурсы за счёт хранения старых событий на более дешёвом железе, и оптимизируем запросы, разгружая продакшен-кластер.
Но этот способ нам не подошёл, потому что два дня — это слишком мало. Тогда мы поделили старые события на две группы. Так как SSD-диски работают медленнее, чем OLTP, но быстрее, чем HDD, мы распределили данные так:
первые 2 дня → OLTP-кластер (максимальная скорость для актуальных данных).
С 3 до 7 дня → SSD-диски (чтобы не перегружать OLTP, но всё ещё быстро).
Данные старше недели → OLAP-кластер на HDD (оптимально для архивного хранения).
Но одной схемы недостаточно. Нужна была ещё одна доработка. Ведь если просто механически распилить профиль между кластерами, никакого профита не будет. Нам нужно было правильно маршрутизировать запросы, чтобы системы знали, где искать данные и не тратили время на ожидание записи на кластер.
Представьте поток входящих событий: новые → новые → новые → старые → новые → новые → новые.
Мы считываем их батчами, ведь обрабатывать по одному — слишком долго, микробатчи эффективнее. Когда посчитали батч, его надо записать. Начинаем разбрасывать данные по кластерам: Новое событие? → Отправляем в OLTP-кластер. Опять новое? → Снова в OLTP-кластер. И снова новое? → Всё ещё в OLTP-кластер.
Возникает ещё проблема: что делать со старыми событиями, чтобы они не нагружали OLTP?
Всё это быстро записывается, пока мы работаем с OLTP-кластером. Но когда доходит дело до старых данных, они уходят на HDD-кластер — и тут начинаются задержки. Новая проблема: MapReduce HDD-кластер ждёт коммита 1000 мс. Мы не можем сдвинуть офсет, пока весь батч не закоммитится. В итоге ждём самого медленного — то есть 1000 мс. Чтобы решить эту проблему, мы решили разбить контуры, чтобы новые и старые события обрабатывались независимо.
На старой схеме появляется ещё один OLAP контур, в который надо как-то записывать. А следить надо за двумя контурами — и OLTP, и OLAP.
Чтобы избавиться от задержек, мы внедрили умный роутинг событий. Событие прилетает в решардеры, из них в мастера. Мастера проверяют ключ в OLTP-контуре: если ключ есть — это свежее событие, мы его сразу отправляем в OLTP. Если ключа нет, значит, оно старое и направляется в очередь старых событий. Очередь старых событий обрабатывается отдельными мастерами, которые работают с HDD-дисками. Главный профит: не ждём commit в HDD. Достаточно закоммитить в очередь то, у которого маленький TTL. Оттуда событие быстро удаляется, но попадает в нужный контур обработки. А дальше всё по знакомой схеме: уведомления (notify) летят в две очереди для новых и старых событий. Подписанные сервисы вычитывают нужное и работают с ними.
В результате у нас больше нет задержек из-за старых событий. OLTP остаётся быстрым и не тратит ресурсы на архивные данные. HDD-данные обрабатываются параллельно, не влияя на свежие запросы. Система работает стабильно и уже успешно крутится в production. Но проблемы с надёжностью остались.
Если мы потеряем стейт, то 100 дней работы рекламы пойдут прахом. Ничего нельзя будет восстановить, никакие воронки не сработают. С этим тоже надо что-то делать — а именно, бэкапы.
Люди делятся на два типа:
Те, кто уже делает бэкапы.
Те, кто ещё не делает бэкапы.
Проблема наших бэкапов в том, что база весит 25 PB, а значит, бэкап тоже весит 25 PB. Но на деле всё ещё хуже.
Расчёт количества бэкапов:
1. В топике лежит не более 1⁄2 суток данных.
2. Релизный цикл – 1 сутки.
3. Время реакции – не больше 1⁄2 суток.
Значит, нужно три бэкапа с частотой в 1⁄2 суток, чтобы иметь возможность восстановиться и накатить все изменения. А три бэкапа по 25 PB — это 75 PB, что ещё хуже по ресурсам.
LSM-деревья
Но мы нашли выход, основанный на LSM-деревьях. Поскольку не все знают, что такое LSM-деревья, объясню на пальцах.
LSM-деревья — это структуры, в которых базы данных хранят информацию. У вас всегда есть кусок свежих данных, который хранится в оперативной памяти. Это данные, которые недавно вставились или изменились. Как только происходит запрос, база данных идёт сначала в RAM, надеясь найти ответ среди «горячих данных». А если там нет, то в отдельные файлики на жестком диске, которые хранятся кусочками. При этом эти данные в каждом куске отсортированы между собой для простоты доступа.
Здесь первый кусочек содержит ключи от 1 до 4, второй — от 6 до 9, третий — от 10 до 23. То есть мы знаем, на какой из кусочков большого массива информации требуется пойти и достать любую запрошенную информацию.
Такое деление на RAM и HDD позволяет не работать всё время с диском, а обращаться к оперативной памяти, чтобы вносить изменения и записывать новые данные. Например, если сделали insert, то добавляем перечисленные ключи в RAM.
Но проблема в том, что RAM конечен. Когда RAM переполняется, это значит, что свежие данные стали слишком большими.
В нашем примере записались ключи: 5, 12, 14. Когда произошло наполнение буфера в RAM, мастер LSM дерева ищет непрерывный диапазон достаточно старых данных, который по своему диапазону совпадает с одним из диапазонов кусочков, хранящихся на HDD. Например, в нашем случае мы видим, что кусочки 11, 12, 14, 18 — непрерывные и хорошо ложатся в старые данные в третий блок. Мы их отделяем от буфера и начинаем записывать на диск, объединяя с одним из кусочков, которые лежали на HDD(мёрджим).
Помёрджив, получаем, что один из блоков изменился, а два других — нет, а ещё у нас уменьшилось то, что мы храним в горячем хранилище в RAM.
Этим мы и воспользовались для уменьшения размера бэкапов. Когда происходит изменение, например, внесение новой информации или апдейты, у нас на диске неизменным остается большой кусок данных. В нашем примере выше изменился только блок третий из трёх.
Поэтому, когда мы делаем три снепшота, то можем бэкапить только изменённые блоки:
Первый снепшот — самый старый. Смотрим, какие блоки изменились по отношению к следующему снепшоту. Если изменились не все, значит, и сохранять их не нужно. Ведь мы их уже сохранили в первой версии. В третьей версии смотрим, что изменилось по отношению ко второму снепшоту и то, что не менялось, нам тоже не нужно хранить второй раз. Я на схеме пометил зеленым те блоки, которые изменились по отношению к предыдущему. В результате, при backup сохраним только первый снепшот и всё, что окрашено зелёным в следующих.
Это даёт нам выигрыш — вместо 75 PB мы храним только 30 PB. На самом деле, ещё меньше, потому что у нас есть и более глубокие оптимизации. Но базовая оптимизация родилась только благодаря тому, что мы знаем, как работает LSM дерево.
Так система с бэкапами стала надёжной и при этом не пожирает огромное количество ресурсов. Мы уже выкатили её на продакшен.
Итого, мы пришли к такому изменению: в прошлой системе мы передавали данные по цепочке:
А теперь мы превратили её в систему вида «звезда», в которой данные сначала отправляются в центральный Key Value Storage, а из него — нужным сервисам.
Эта схема позволяет прокидывать данные между событиями минуя промежуточные этапы. То есть, если нам нужен referer антифрода, мы просто закидываем его в профиль, а из профиля отправляем в сам антифрод. Сервисы, отвечающие за показы и клики, эти данные не используют и ничего с ними не делают.
Система готова. Выводы
Мы поняли, что архитектура «звезда»:
Дешевле
Теперь у нас возникает меньше инцидентов, связанных с ошибками в статистике и витринах данных. Когда мы запустили новую архитектуру, она позволила уменьшить количество данных, число ошибок в данных, статистике, работе аналитиков с данными. Ведь часть бизнес-логики мы поместили в сам обработчик стейта. Теперь она применяется везде единообразно, без расхождений, by design.
Меньше ресурсов тратится на MapReduce.
Меньше кода. Новая схема позволяет делать многое в формате less code. Если вы хотите прокинуть поле, то у нас сделаны конфигурации, в которых написано, в каком направлении какие поля доставляются.
Меньше ресурсов на поддержку за счёт единообразия. Поскольку всё консистентно, то если потребители используют старый формат, например, TSKV, мы просим переехать на Protobuf, который используют почти все. Мы просто сообщаем, что больше не поддерживаем TSKV. Такой подход даёт нам уменьшение строк кода, меньше мест, где можно ошибиться, и обеспечивает простоту поддержки.
Проще
Там, где раньше надо было писать код, сейчас достаточно аннотации в .proto-файле. Если нужно отправить специфическое событие трём сервисам, то достаточно в .proto-файле написать, что нужно добавить поле в антифрод и, например, биллинг. Наша система видит .proto-файл и отправляет всё туда, куда нужно.
Стимулирует переходить на единый формат сообщений.
Разделены инфраструктура и бизнес-логика. Бизнес-логика выделена в отдельную библиотеку. Есть инфраструктура, управляющая всеми стейтами. А есть библиотека под названием продуктовая матрица, в которой записана бизнес-логика.
Большая переиспользуемость кода. Поскольку у вас один стейт, то вокруг него написано много библиотек. Если вы хотите что-то сделать, то наверняка уже кто-то делал это до вас, просто переиспользуйте.
Надёжнее
Потеря одного события не ведёт к потере всей цепочки. Если внезапно потерялось событие показа, это не катастрофа. Да, сервис рекламы показа не увидит, но антифрод всё равно получит нужные данные о клике. А если из клика произошла конверсия, то метрика спишет деньги с бюджета рекламодателя и отобразит это у него в статистике. Да, возможно, ML на этом клике не обучится. Но наш сервис всё равно и денег заработает, и рекламодатель увидит, с какого источника поступают клики.
Возможность делать бэкапы. В прошлой схеме делать бэкапы было в принципе невозможно. Речь про закодированные access.log редиректов url’ов. А сейчас у нас есть бэкапы, на которые мы можем откатиться в случае инцидента.
Можно тестировать всю цепочку сообщений. Тестировать стало проще, потому что в старой схеме нужно было писать интеграционные тесты по куче систем, а в новой — система одна, и она управляет всеми событиями. Вы можете написать для неё самые разные тесты.
Переиспользуемость кода, одна точка применения, разнесение бизнес-логики. Бизнес-логику иногда пишут аналитики, иногда даже джуны, но инфраструктуре они не смогут навредить.
23-24 июня в Санкт-Петербурге снова поговорим о распределенных системах, оптимизации и обработке данных, а так же о их хранилищах. Программа профессиональной конференции для разработчиков высоконагруженных систем и все подробности Saint HighLoad++ 2025 на официальном сайте.