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

Наш Feature Store — Avalon — возник в момент, когда понадобилось масштабируемое и производительное хранилище с низкой задержкой, в котором можно структурировать признаки по иерархии «каталог/файл», получать быстрый доступ к ним из рантайма, автоматически отслеживать актуальность данных и контролировать жизненный цикл каждого признака. Роль СУБД для системы выполняет YDB, что позволяет достичь высокой отказоустойчивости и горизонтального масштабирования.

Всем привет! Меня зовут Паша, я руковожу группой разработки технологий эффективности Такси. В этой статье я расскажу, как мы проектировали и строили Avalon, какие вызовы пришлось решать команде по мере роста нагрузок и аудитории, почему прежние подходы перестали соответствовать задачам современного продуктового анализа и как в результате получился удобный и надёжный Feature Store для множества бизнес-сценариев.

С чего всё началось

Итак, начинается понедельник. Я открываю Яндекс Трекер и вижу такую задачу:

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

К задаче прилагаются характеристики: RPS, объёмы данных и так далее. Всего признаков — 2 млрд, из которых ежедневно обновляются около 2 млн. Очевидно, им нужно место, где всё это хранить. Причём задача общая: такая функциональность может пригодиться и другим сервисам.

У команды уже была структура данных и тип ключа, по которому всё строится, а наша задача состояла в том, чтобы обеспечить консистентный доступ к этим данным в рантайме. Нам принесли YT-таблицу, в которой примерно 1,7 млрд записей. Нужно было сделать так, чтобы обращаться в рантайме к этим данным можно было с большим RPS, а при необходимости — кешировать их и быстро получать ответ, проставлен атрибут пользователю или нет.

Далее на основе признака «есть атрибут/нет атрибута» сервис решает, что показать пользователю. В более общем смысле наша задача — дать сервисам универсальный способ получать обновляемые данные с низкой latency, без участия разработчиков или других команд.

Разметка аудитории — это типовая задача, и до Avalon в Городских сервисах уже были другие решения. Его предшественник — сервис с PostgreSQL под капотом для хранения бинарных признаков. Он задумывался как способ быстро и экспериментально размечать аудиторию. И на основе полученных результатов заказывать доработки в нужных сервисах.

Сервис быстро стал популярным. Однако пользователи — аналитики и разработчики — столкнулись с проблемами:

• Пользователи хотели загружать всё больше и больше данных. А PostgreSQL нужно масштабировать вручную.

• Признаки в сервисе имели бинарную природу: отвечали на вопрос только «да» или «нет». Но если мы хотим узнать, например, количество поездок водителя за всё время, то такой подход уже не поможет.

Так появился Avalon — инструмент, позволяющий удобно хранить, структурировать и быстро получать самые разные пользовательские признаки. Название Avalon выбрал разработчик, который вдохновляется кельтской мифологией. Согласно современному толкованию, миф об Авалоне — отголосок сказаний об «островах блаженных». В общем, это что-то похожее на родные нашему уху «кисельные реки и молочные берега», отсылающие к тем же самым образам комфорта и изобилия, которые мы вкладывали в саму идею системы.

Требования к системе

При разработке Avalon у нас было несколько важных требований:

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

• Должна быть возможность хранить не только бинарные признаки вроде «пользователь с атрибутом или нет», но и произвольные признаки, например количество поездок, совершенных пользователем за всё время.

• Должен быть удобный реестр признаков: понимать, кто ими пользуется в рантайме, кому они нужны и сколько времени каждый признак должен жить. Если признак нужно удалить, это должно происходить безопасно и предсказуемо.

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

curl --env testing avalon-proxy \
  -v avalon-proxy.somewhere/v1/fetch \
  -X POST -H "Content-Type: application/json" \
  -d '{"id": "123", "features": ["/service-general/{}/features/with_attribute"]}'

{
  "id": "123",
  "features": {
    "/service-general/{}/features/with_attribute": {
      "data": {
        "@type":
        "type.googleapis.com/NUgc.NSchema.NAvalon.TGgeneralFeature",
        "name": "with_attribute"
      }
    }
  }
}

Архитектура Avalon: общее устройство

Точка входа в систему — сервис Avalon Proxy. Мы строили систему так, чтобы данные хранились в абстрактных хранилищах, а для пользователя это было скрыто за прокси.

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

Ядром системы выступает UGCDB Framework — это key-value хранилище, которое задаёт определённые правила работы с данными. UGCDB позволяет описывать логическую схему данных. Если у одной записи может быть несколько подзаписей (листов у узла), это называется таблицей в контексте UGCDB Backend. Внутри таблицы могут находиться другие таблицы — как папки внутри папок. Если у записи может быть только одно значение, это колонка. Эти понятия и ограничения задал сам фреймворк, который был написан до нас.

Под капотом — YDB. Для нас это удобно прежде всего тем, что YDB обеспечивает шардирование «из коробки». Кроме того, в системе есть автоматическая генерация API для доступа к данным. Данные у нас представлены в виде логической таблицы, и это тоже даёт преимущества. Мы можем описать структуру данных с таблицами и колонками, как она выглядит логически, но под капотом это одна физическая таблица. Благодаря этому таблицу удобно шардировать.

И ещё одно преимущество — наличие CDC (Change Data Capture). Так мы можем отслеживать изменения данных с течением времени, что полезно, например, для финансовых данных.

Архитектура Avalon: хранение данных

В физической таблице перечислены следующие поля:

В исходниках нашей задачи — 2 млрд записей. Для них мы описали логическую схему данных следующим образом:

• есть таблица ServiceGeneral;

• внутри неё — таблица features;

• внутри features лежат данные.

message TRoot {
  repeated TGeneralFeatureStorage ServiceGeneral = 1;
  option (schema) = {
    Table: {
      Name: "service-general"
      Field: "ServiceGeneral"
      Key: "Id"
    }
  }
}

message TGeneralFeatureStorage {
  string Id = 1 [json_name = "id"];
  repeated TGeneralFeature Features = 2 [json_name = "features"];

  option (schema) = {
    Table: {
      Name: "features"
      Field: "Features"
      Key: "Name"
    }
  };
}

message TGeneralFeature {
  string Name = 1 [json_name = "name"];
  google.protobuf.Value Payload = 2 [json_name = "payload"];
}

Если вспомнить, в YT-таблице, которую приносили коллеги, прямо указано:

name = with_attribute

Таким образом мы загрузим в поле Name структуры TGeneralFeature значение with_attribute для каждого идентификатора из YT-таблицы.

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

Теперь посмотрим, как логическая схема соотносится с тем, как мы получаем данные. Для запроса вызывается endpoint

avalon-proxy.somewhere/v1/fetch

В качестве фичи пользователь передаёт путь:

/service-general/123/features/with_attribute

Это признак пользователя с атрибутом в домене service-general. Мы называем доменом тот префикс, который указан в начале — здесь это service-general. Таким образом, можно получить точное значение признака для пользователя с идентификатором 123.

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

/service-general/123/features

В ответ возвращаются все признаки, которые есть у пользователя.

Архитектура Avalon: работа с СУБД

Теперь поговорим об обновлении и получении данных в YDB. Доступны следующие операции:

• точечное чтение и range-based-чтение,

• точечная запись,

• модифицирующие запросы с сompare-and-swap,

• точечное удаление,

• batch-запросы.

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

DECLARE $shard_id AS Uint32; // CityHash(root level id -> 123)
DECLARE $key_id AS String; // "/service-general/123/features/with_attribute"
SELECT ShardId, Key, ValueProto
FROM `data`
WHERE ShardId = $shard_id AND Key = $key;

Кроме того, можно получить все признаки, а не только with_attribute. Для этого в запросе используется range-based-чтение:

DECLARE $shard_id AS Uint32;
DECLARE $start_key AS String;
DECLARE $end_key AS String;
SELECT ShardId, Key, ValueProto 
FROM `data`
WHERE ShardId = $shard_id AND Key >= $start_key AND Key < $end_key;

• начало диапазона — "/service-general/123/features/ ", где " " — минимальный ASCII символ, разрешенный для использования в названиях.

• конец диапазона — "/service-general/123/features/~", где "~" — последний ASCII-символ, разрешенный для использования в названиях.

Так можно получить все признаки разом.

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

DECLARE $shard_id AS Uint32;
DECLARE $key_list AS List<Tuple<Uint32?,String?>>;
SELECT ShardId, Key, ValueProto
FROM `data`
WHERE (ShardId, Key) IN $key_list;

Модификации выполняются обычным upsert.

// Запись
DECLARE $shard_id AS Uint32;
DECLARE $key AS String;
DECLARE $value AS String;
...
UPSERT INTO `data` (ShardId, Key, ValueProto, ...)
VALUES ($shard_id, $key, $value, ...);
// для IfNotExist и IfNotEqual используются интерактивные транзакции
// логика их обработки реализована на уровне приложения
// Удаление
DECLARE $shard_id AS Uint32;
DECLARE $key AS String;
DELETE FROM `data` WHERE ShardId = $shard_id AND Key = $key;

Архитектура Avalon: загрузка и выдача данных

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

1. Читается внутреннее состояние системы.

2. Читается состояние таблицы, полученной в данный момент.

3. Внешний сервис считает разницу между ними.

4. Подготавливается таблица для загрузки.

5. Скрипт на C++ загружает данные в UGCDB Backend.

6. Backend записывает данные в YDB.

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

После загрузки данных пользователь должен иметь возможность их получить. Для этого написаны два сервиса: Avalon Proxy и Avalon Data Processor. Они решают задачу абстрагирования пользователя от внутреннего устройства хранилища. Кроме того, в них реализован ACL: определено, кто может потреблять конкретный признак, с каким RPS и на каком объёме железа, в зависимости от договорённости с командой.

В Avalon Proxy реализован усечённый API. Если UGCDB Backend позволяет произвольный доступ к данным, то в Proxy всё строже: здесь запросы только с одним верхеуровневым идентификатором, чтобы гарантировать одношардовость батчей. Это нужно, чтобы контролировать нагрузку и предсказуемо оценивать SLA.

Avalon Data Processor решает задачу ACL — управления доступом к признакам. В отличие от сервиса предшественника, где было неясно, кто что использует, здесь все пользователи известны и ограничения задаются явно. ACL определяет, какие сервисы могут потреблять конкретные признаки: например, service_loyalty получает один набор фич, а service_perks — другой. Avalon Proxy на этапе запроса проверяет права и ограничивает нагрузку.

{
  "service-loyalty": {
    "/service-general/{}/features": {
    }
  },
  "service-perks": {
    "/service-general/{}/features": {},
    "/passport-uids/{}/service-payment-methods": {},
    "/passport-uids/{}/service-prev-business-purchases": {},
    "/passport-uids/{}/personal-goals": {}
  }
}

Так замыкается контур: данные из YT-таблиц загружаются в систему, а клиент получает их через Avalon Proxy.

Интеграция оказалась быстрой: на стороне сервиса это всего около двадцати строк кода на C++.

std::vector<std::string> AvalonProxyService::GetAvalonGeneralFeatures() const {
auto response_opt = GetGeneralFeaturesData(avalon_get_features_task_);
if (!response_opt) {
  return {};
}
auto response = response_opt.value();
std::vector<std::string> result; result.reserve(response.GetFeatures().size());
for (const auto& feature : response.GetFeatures()) {
   if (feature.GetName() == kServiceGenderPromo) {
       continue;
   }
    if (feature.HasPayload()) {
        const auto& payload = feature.GetPayload();
        if (payload.has_struct_value()) {
           const auto& protoMap = payload.struct_value().fields();
           const auto* suffixes = MapFindPtr(protoMap, "suffixes");
           if (suffixes) {
              if (suffixes->has_list_value()) {
                 for (const auto& suffix : suffixes->list_value().values()) {
                     if (suffix.has_string_value()) {
                        result.push_back(feature.GetName() + '_' +             suffix.string_value());
                        boost::to_upper(result.back());
                 } 
               }
             }
             continue;
           }
         }
       }
     }
    result.push_back(feature.GetName());
    boost::to_upper(result.back());
   }
   return result
}

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

Архитектура Avalon: Change Data Capture

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

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

Возникает задача: нужно в реальном времени захватывать каждое изменение (INSERT, UPDATE, DELETE) в основной таблице, обогащать его метаданными и доставлять в виде потока событий для дальнейшей обработки. Именно эту задачу и решает CDC, который в нашей имплементации называется Feeder.

Поток данных выглядит так:

• Перед применением изменения приложение (UGCDB Backend) читает актуальное состояние записи, формирует запись для истории изменений и затем вносит ее вместе с основным изменением в таблицу YDB.

• Feeder читает специальные записи истории изменений и перекладывает их в YDB Topics.

• Подписчики топика (другие сервисы) потребляют события.

// Упрощённая схема сообщения
message TFeedRecord {
  string Key = 1;
  TRecordMeta OldMeta = 2;
  google.protobuf.Any OldValue = 3;
  TRecordMeta NewMeta = 4;
  google.protobuf.Any NewValue = 5;
  TRequestMeta RequestMeta = 6; // Контекст запроса
}

В случае горизонтального масштабирования количество экземпляров Feeder всегда будет равно количеству шардов в исходной таблице YDB. Каждый Feeder закреплён за своим шардом. В рамках одного шарда Feeder гарантирует, что события будут отправлены в топик ровно в той последовательности, в которой изменения произошли в БД. Это критически важно для корректного восстановления состояния. Для этого используются монотонные счётчики YDB updateEpoch и updateNumber, которые увеличиваются при каждой новой записи.

Важно отметить, что после успешной записи в топик, Feeder удаляет эту запись из базы. Если в этот момент произошла ошибка и запись не удалилась, то при последующем запуске YDB topics на своей стороне отфильтрует её как уже обработанную. Таким образом у Feeder есть гарантии доставки «ровно один раз».

А для наших пользователей всё это выглядит простым добавлением атрибутов в protobuf-схеме.

message TRoot {
  repeated TGeneralFeatureStorage DiscountsGeneral = 1 [(topic) = "discounts-general-json"];

  option (schema) = {
    Table: {
      Name: "discounts-general"
      Field: "DiscountsGeneral"
      Key: "Id"
    }
  };
}

Результаты

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

Сейчас в системе хранится около 3 ТБ данных и 6,5 млрд ключей. Всего используется 2048 шардов YDB. Нагрузка — 100 тысяч RPS на чтение, 30 тысяч RPS на запись, при этом задержка в P95 — около 5 мс.

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

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