#ksql

Сегодня рассмотрим, как дата-инженеры маркетплейса Whatnot масштабировали потоковую обработку данных с помощью Apache Kafka, изменив свои ETL-процессы и реализовав на этой распределенной платформе шину событий для анализа пользовательского поведения c ksqlDB и Rockset.Постановка задачи: события пользовательского поведения в WhatnotWhatnot – это маркетплейс, пользователи которого могут покупать и продавать товары в рамках публичных аукционов, а также общаться между собой. Чтобы улучшить пользовательский опыт, Whatnot использует машинное обучение, предлагая своим клиентам товары, которые могут их заинтересовать.ETL-стратегия компании заключалась в обработке пакетов журналов сбора данных об изменениях из транзакционных систем приложения в корпоративное хранилище данных. Приняв данные, CLI-инструмент dbt с открытым исходным кодом, который помогает дата-аналитикам и инженерам эффективно преобразовывать данные в хранилищах, переводит эти логи в удобные для пользователя таблицы. Сформированные таблицы далее используются несколькими нижестоящими приложениями.Однако, транзакционные системы не очень хорошо подходят для анализа данных и наоборот. Например, в начале мая 2022 года была выпущена функцию под названием «Реакции», благодаря которой пользователи могут выразить свое отношение прямо в реальном времени на видеотрансляции, чтобы запустить анимацию, наложенную на видеопоток.При реализации этой функции было решено не хранить каждую реакцию в транзакционной системе, поскольку они чрезвычайно требовательны к записи. Реакции срабатывают, когда любой пользователь в любой прямой трансляции нажимает кнопку реакции. Сохранение каждой отдельной реакции сильно увеличит объем операций записи и, следовательно, общую нагрузку, на уровень хранения.Хотя это решение было правильным для отдельной фичи, оно создало проблему, превратив ценную для анализа информацию в «темные данные», которые существуют, но нигде не собираются в транзакционных системах. Поэтому дата-инженеры Whatnot решили отделить реализацию хранилища фичи от передачи данных об ее использовании для анализа и моделирования. Для этого пришлось реализовать шину событий на Apache Kafka, что мы и рассмотрим далее.Шина событий на Apache KafkaApache Kafka как шина событий была запущена в Confluent Cloud. Три внутренние сервисы Whatnot отправляют события на шину:- основной серверный сервис, который используется для критически важных медленно движущихся данных, таких как заказы и отгрузки;- сервис в реальном времени для оперативных быстрых данных, таких как чаты и аукционы;- сервис ранжирования, нужный для управления лентой пользовательских интересов.Каждый из этих сервисов является первичным продюсером событий и работает со своим собственным уникальным набором ограничений. Одним из главных нефункциональных требований к этим продюсерам является их высокая надежность. Чтобы гарантировать их корректное поведение, был написан ряд модульных тестов, которые запускаются каждый раз, когда регистрируются любые изменения в коде. Для реализации шины событий был запущен локальный кластер Apache Kafka в рамках CI/CD-процессов с использованием Docker. Это позволяет приложению отправлять события в тестах способом, аналогичным реальному производству, что дает возможность писать проводить тестирование «черного ящика».Эти типы тестов помогают проверить структуру события, подтверждая, что оно (де)сериализуемо и содержит соответствующие поля, а также понять корректность самого события, убедившись, что оно содержит конкретный фрагмент данных. Можно изменить детали реализации производства событий без необходимости изменять тесты.При разработке продюсеров приходится выбирать подходящие схемы данных для разных типов событий. Рассмотрев свои варианты использования, дата-инженеры Whatnot остановились на двух схемах событий, которые охватывают большинство их типов:- события пользовательского поведения запускаются после того, как пользователи совершают определенные действия, такие как общение в чате или нажатие кнопки реакции;- события продукта в прямой трансляции запускаются, когда продукты в прямой трансляции претерпевают соответствующее изменение состояния, например, когда начинается или заканчивается аукцион для продукта.Эти альтернативные схемы позволяют просто и безопасно добавлять новые события без создания новых топиков Apache Kafka, разработки дополнительных конвейеров данных, моделей хранилища и пр. Достаточно лишь зарегистрировать изменение в схеме, написать код и тест.Вопросы определения структуры данных, т.е схемы, неразрывно связаны с форматом сериализации сообщений при их передаче по сети. Apache Kafka поддерживает несколько форматов. В частности, двоичные Protocol Buffers и AVRO имеют лучшую производительность и обратную совместимость, по сравнению с человеко-читаемым JSON. Однако, дата-инженеры Whatnot решили использовать JSON из-за его простоты и наглядности. Впрочем, чтобы в будущем без проблем изменить формат сериализации, продюсеры событий были реализованы таким образом, что детали сериализации абстрагированы от вызывающих их объектов. Отделение деталей сериализации сообщений от кода, генерирующего события, дает возможность в будущем изменить формат сериализации, заменив объемный JSON на легкий Protobuf или AVRO.Для потоковой обработки событий используется ksqlDB и Rockset – аналитическая платформа на базе встроенного key-value хранилища RocksDB, которое используют приложения Kafka Streams. Аналитическая платформа Rockset обеспечивает поиск, агрегацию и соединение массивных полуструктурированных данных в режиме реального времени без дополнительной нагрузки, автоматически индексируя структурированные, полуструктурированные, географические и временные данные для поиска в реальном времени и масштабной аналитики. Подробно об этом мы писали здесь.Rockset применяется для простых операций типа JOIN-соединений и свертки, а также для хранения и обслуживания результатов потоковой обработки, чтобы их можно было использовать в приложении. Для более интенсивной потоковой обработки, например, агрегации скользящих окон, используется ksqlDB, что позволяет отправлять метрики снова в Rockset для их последующего анализа и повторного использования.Комбинация ksqlDB и Rockset позволяет рассчитывать метрики в реальном времени на основе событий и возвращать их в приложение почти сразу же. В результате этого аналитические и транзакционные сервисы Whatnot могут ежедневно принимать десятки миллионов событий о реакциях в схеме, удобной для анализа и моделирования, причем реализация событий полностью отделена от функции их хранения.Больше подробностей про администрирование и эксплуатацию Apache Kafka в системах аналитики больших данных вы узнаете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:- Apache Kafka для инженеров данных- Администрирование кластера KafkaИсточники1. https://medium.com/whatnot-engineering/scaling-our-data-stack-with-kafka-and-real-time-stream-processing-56554dcbb0fc2. https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html