Содержание статьи
Постановка задачи
Задача была немного необычной для нашего DWH на основе Arenadata DB / Greenplum. А именно в DWH требовалось организовать приём событий пользовательского поведения из RabbitMQ и дополнительно предоставить команде e-commerce доступ к данным с задержкой не более пяти секунд. Под задержкой далее понимается интервал между публикацией сообщения в RabbitMQ и доступностью соответствующей записи для запроса в СУБД. Была выбран следующая схема обработки данных — потоковая запись из RabbitMQ в ClickHouse и минибатчинг из ClickHouse в Arenadata DB. В статье рассматривается первая часть схемы — интеграция RabbitMQ и ClickHouse. В самой статье рассматриваются несколько хинтов, которые помогли ускорить обработку данных и корректно обрабатывать сами сообщения (особенно это касается крзин).
В исходных именах таблиц и представлений используется термин customer tracking. В данной статье он обозначает события взаимодействия пользователя с платформой интернет-магазина: просмотры и переходы, поисковые запросы, операции с корзиной и действия в мобильном приложении. То есть термин clickstream events.
Сразу было принято решение записывать в отдельные таблицы:
-
таблица с исходными сообщения в виде JSON-логов — для диагностики, аудита преобразований и повторного разбора;
-
таблица с типизированными колонками — для аналитических запросов без регулярного парсинга JSON.
П.1 обеспечивает независимость от команды разработки платформы интернет-магазина и защищает (во всяком случае частично) обработчик очереди от одностороннего изменения схемы сообщений.
Ограничение в пять секунд относится к конвейеру целиком. В представленной конфигурации значение rabbitmq_flush_interval_ms = 4000 резервирует до четырёх секунд для формирования пакета на стороне RabbitMQ-движка. Однако само по себе оно не гарантирует соблюдение end-to-end SLA: в измерение также входят публикация в брокер, доставка сообщения, обработка Materialized View и вставка в целевую таблицу. Поэтому фактическое соблюдение требования следует контролировать по метрикам задержки между временем публикации и временем доступности данных в ClickHouse.
Результаты нагрузочного тестирования
На испытательном стенде конвейер сохранял работоспособность при входной нагрузке до 1 000 сообщений в секунду. Задержка измерялась от публикации сообщения до доступности соответствующей записи в ClickHouse.
Распределение задержки при этой нагрузке имело следующий вид:
-
72% сообщений стали доступны менее чем за 5 секунд;
-
100% сообщений стали доступны менее чем за 7 секунд.
Таким образом, в рамках проведённого испытания задержка не превысила семи секунд, тем не менее заказчик посчитал требования выполнеными.
Результаты относятся к конкретной конфигурации стенда и профилю нагрузки, которые я здесь не привожу. В статье разбирается подход и некоторые неочевидные решения, которые позволили достичь требуемых результатов (о них ниже).
В схеме rabbitmq_queue_base реализован набор однотипных конвейеров для событий веб-трекинга, поиска, мобильного приложения и корзины. Каждый конвейер разделяет приём сообщений, хранение исходного содержимого и формирование структурированного слоя. Такое разделение позволяет независимо управлять надёжностью приёма и правилами преобразования данных.
Состав и взаимосвязь компонентов
RabbitMQ exchange │ ▼ *_queue ENGINE = RabbitMQ │ (raw_message String) ▼ *_mv MATERIALIZED VIEW │ TO *_data ▼ *_data ReplicatedMergeTree │ (event_date, raw_message) ▼ *_mv_ext_json MATERIALIZED VIEW + JSONExtract │ TO *_data_extracted ▼ *_data_extracted ReplicatedMergeTree (типизированные поля, TTL 90 дней)
Для потока корзины предусмотрен дополнительный этап: элементы массива entries преобразуются в строки отдельной таблицы customer_cart_entry_data_extracted.
В схеме определены следующие конвейеры:
Канал событий
Exchange / routing key
Таблица исходных сообщений
Таблица структурированных данных
Веб-трекинг
customer-tracking
customer_tracking_data
customer_tracking_data_extracted
Поисковые события
customer-search-tracking
customer_search_tracking_data
customer_search_tracking_data_extracted
Мобильный трекинг
customer-mobile-tracking
customer_mobile_tracking_data
customer_mobile_tracking_data_extracted
События корзины
customer-cart-tracking
customer_cart_tracking_data
customer_cart_tracking_data_extracted
Состояние корзины
customer-cart
customer_cart_data
customer_cart_data_extracted и данные позиций
Ниже базовый конвейер рассмотрен на примере customer_tracking; последующие разделы фиксируют особенности остальных потоков и обработку состава корзины.
1. Приём сообщений: таблица на движке RabbitMQ
Таблица с ENGINE = RabbitMQ является точкой интеграции ClickHouse с брокером. Движок получает сообщения из заданного exchange и предоставляет их как строки с телом сообщения в колонке raw_message.
CREATE TABLE rabbitmq_queue_base.customer_tracking_queue(raw_message String) ENGINE = RabbitMQ SETTINGS rabbitmq_host_port = ‘rabbitmq.example.com:5672’, rabbitmq_exchange_name = ‘customer-tracking-exchange’, rabbitmq_exchange_type = ‘direct’, rabbitmq_routing_key_list = ‘customer-tracking’, rabbitmq_queue_base = ‘customer-tracking’, rabbitmq_queue_consume = true, rabbitmq_format = ‘JSONAsString’, rabbitmq_num_consumers = 4, rabbitmq_max_block_size = 50000, rabbitmq_flush_interval_ms = 4000, rabbitmq_vhost = ‘/app’;
Существенные параметры конфигурации имеют следующее назначение:
-
rabbitmq_format = ‘JSONAsString’ сохраняет payload без предварительной интерпретации. Правила извлечения полей могут изменяться независимо от настроек консьюмера.
-
rabbitmq_queue_consume = true включает режим потребления сообщений.
-
rabbitmq_num_consumers = 4 задаёт число параллельных потребителей.
-
rabbitmq_max_block_size и rabbitmq_flush_interval_ms определяют условия формирования пакета: не более 50 000 строк либо истечение четырёх секунд.
-
rabbitmq_exchange_name, rabbitmq_routing_key_list, rabbitmq_queue_base и rabbitmq_vhost идентифицируют источник сообщений в RabbitMQ.
Во всех таблицах вида *_queue используется одинаковая конфигурационная модель; различаются идентификаторы exchange, routing key и очереди.
2. Слой исходных сообщений
Таблица хранения исходных сообщений
CREATE TABLE rabbitmq_queue_base.customer_tracking_data ( event_date DateTime DEFAULT now(), raw_message String ) ENGINE = ReplicatedMergeTree( ‘/clickhouse/tables/{shard}/customer_tracking_data’, ‘{replica}’ ) ORDER BY tuple() TTL event_date + toIntervalDay(90) SETTINGS index_granularity = 8192;
-
event_date фиксирует время вставки строки в ClickHouse.
-
raw_message содержит неизменённое представление payload.
-
ORDER BY tuple() соответствует журналу последовательных вставок, для которого не предполагается выраженный ключ доступа. В customer_cart_tracking_data используется сортировка по event_date.
-
TTL продолжительностью 90 дней ограничивает срок хранения исходных сообщений. В течение этого периода возможно повторное преобразование данных при изменении логики разбора.
Материализованное представление для записи исходных сообщений
CREATE MATERIALIZED VIEW rabbitmq_queue_base.customer_tracking_mv TO rabbitmq_queue_base.customer_tracking_data ( event_date DateTime, raw_message String ) AS SELECT raw_message FROM rabbitmq_queue_base.customer_tracking_queue;
Материализованное представление не образует самостоятельного хранилища. При вставке блоков в исходную таблицу *_queue выполняется его запрос, а результат записывается в таблицу, указанную после TO. Поскольку в выражении SELECT передаётся только raw_message, значение event_date формируется выражением DEFAULT now() в целевой таблице.
Аналогичную функцию выполняют представления customer_search_tracking_mv, customer_mobile_tracking_mv, customer_cart_tracking_mv и customer_cart_mv. Материализованное представление реагирует на новые вставки в исходную таблицу; оно не обрабатывает строки, уже находившиеся в ней к моменту создания представления.
3. Структурирование JSON-сообщений
Таблица структурированных событий
CREATE TABLE rabbitmq_queue_base.customer_tracking_data_extracted ( event_date DateTime DEFAULT now(), pk Nullable(Decimal(20, 0)), systemid Nullable(String), resultscount Nullable(Decimal(20, 0)), listingmodeparam Nullable(String), createdts DateTime, cartcode Nullable(String), uiel Nullable(String), paymenttype Nullable(String), userid Nullable(String), itemprice Nullable(Decimal(30, 8)), responsecode Nullable(String), ajax Nullable(UInt8), url Nullable(String), gaid Nullable(String), salesapplication Nullable(String), salesapplicationid Nullable(String), ip Nullable(String), referer Nullable(String), subtotal Nullable(Decimal(30, 8)), customerid Nullable(String), companyid Nullable(String), sortparam Nullable(String), useragent Nullable(String), domain Nullable(String), localip Nullable(String), deliverymode Nullable(String), sessionid Nullable(String), basestore Nullable(String), productfamilymarker Nullable(UInt8), origuserid Nullable(Decimal(20, 0)), minquantity Nullable(Decimal(20, 0)), proxyname Nullable(String), listingchanges Nullable(String) ) ENGINE = ReplicatedMergeTree( ‘/clickhouse/tables/{shard}/customer_tracking_data_extracted’, ‘{replica}’ ) PARTITION BY toYYYYMM(event_date) ORDER BY event_date TTL event_date + toIntervalDay(90) SETTINGS index_granularity = 8192;
Схема таблиц вида *_data_extracted основана на следующих принципах:
-
партиционирование PARTITION BY toYYYYMM(event_date) группирует данные по месяцу загрузки и упрощает удаление устаревших партиций;
-
ключ ORDER BY event_date ориентирован на преобладающий сценарий выборок по временному интервалу;
-
тип Nullable используется для полей, отсутствие которых допустимо в сообщении;
-
createdts отражает время, переданное источником, тогда как event_date — время загрузки в ClickHouse.
Материализованное представление для извлечения полей
Преобразование реализовано одним вызовом JSONExtract(raw_message, ‘Tuple(…)’). Результат извлечения имеет тип Tuple и далее распределяется по колонкам целевой таблицы посредством обращений parsed.N.
CREATE MATERIALIZED VIEW rabbitmq_queue_base.customer_tracking_mv_ext_json TO rabbitmq_queue_base.customer_tracking_data_extracted AS SELECT event_date, parsed.1 AS pk, parsed.2 AS systemid, parsed.4 AS resultscount, parsed.6 AS listingmodeparam, parseDateTimeBestEffort(parsed.7) AS createdts, parsed.8 AS cartcode, parsed.9 AS uiel, parsed.10 AS paymenttype, parsed.11 AS userid, parsed.12 AS itemprice, parsed.13 AS responsecode, CAST(parsed.14, ‘Nullable(UInt8)’) AS ajax, parsed.15 AS url, — … остальные поля … parsed.5 AS minquantity, parsed.33 AS proxyname, parsed.3 AS listingchanges FROM ( SELECT event_date, JSONExtract( raw_message, ‘Tuple( Nullable(Decimal(20, 0)), Nullable(String), Nullable(String), Nullable(Decimal(20, 0)), Nullable(Decimal(20, 0)), Nullable(String), String, Nullable(String), /* … типы в порядке ключей JSON … */ Nullable(String) )’ ) AS parsed FROM rabbitmq_queue_base.customer_tracking_data AS c ) AS raw_data;
При поддержке такого представления необходимо учитывать три аспекта:
-
Извлечение выполняется позиционно. Соответствие parsed.N полю сообщения определяется декларацией типа Tuple(…) и должно быть зафиксировано в контракте источника. Порядок колонок целевой таблицы не является основанием для такого соответствия.
Здесь нужно отметить, что именно позиционное извлечение позволило достичь целевых показателей при нагрузочном тестировании. -
Строковые представления дат преобразуются функциями parseDateTimeBestEffort и parseDateTimeBestEffortOrNull. Второй вариант применяется, когда отсутствие или некорректное значение даты считается допустимым.
-
Логические значения из JSON приводятся к типам колонок явно: например, к Nullable(UInt8) либо Nullable(Decimal(1, 0)).
4. Специфика предметных потоков
Потоки трекинга используют общий набор атрибутов контекста: идентификаторы пользователя и сессии, URL, IP-адрес, referer и другие параметры клиентского взаимодействия. Специализированные потоки дополняют этот набор атрибутами соответствующей предметной области.
Поисковые события: customer_search_tracking_*
В таблице структурированных данных дополнительно представлены finalsearchtext, searchtext, elevate, assortmentgroup, scom, scomv, cataloglimit, queryexclude и basestoreid. Эти атрибуты описывают параметры и результат выполнения поискового запроса.
Мобильный трекинг: customer_mobile_tracking_*
Специфическим атрибутом является actiontype, характеризующий действие в мобильном приложении. Для minquantity установлен тип Nullable(Int32), в отличие от Decimal-представления в ряде других потоков.
События корзины: customer_cart_tracking_*
Помимо общих полей, поток содержит qid, referercartaddedpage, type, productcode, productadditiontype, itemcost и count. Они позволяют описать контекст изменения корзины и товарную позицию, к которой относится действие.
Состояние корзины: customer_cart_*
Данный поток описывает состояние корзины, а не отдельное пользовательское действие. Наряду с реквизитами корзины в сообщении передаётся массив её позиций.
CREATE TABLE rabbitmq_queue_base.customer_cart_data_extracted ( event_date DateTime DEFAULT now(), localpk Nullable(Decimal(20, 0)), modifiedts Nullable(DateTime), createdts DateTime, totalprice Nullable(Decimal(30, 8)), code Nullable(String), paymentmodecode Nullable(String), user Nullable(String), unit Nullable(String), phonenumberextid Nullable(String), store Nullable(String), payername Nullable(String), eliteorderid Nullable(String), deliverycost Nullable(Decimal(30, 8)), deliverycostinelite Nullable(Decimal(30, 8)), topvisitedstep Nullable(Decimal(20, 0)), topvisitedstepdate Nullable(DateTime), deliveryaddressextid Nullable(String), deliverymodecode Nullable(String), entries Array(String) ) ENGINE = ReplicatedMergeTree(…) PARTITION BY toYYYYMM(event_date) ORDER BY event_date TTL event_date + toIntervalDay(90);
Колонка entries Array(String) сохраняет позиции корзины в виде массива JSON-фрагментов. Это обеспечивает сохранность исходной структуры одновременно с формированием отдельной нормализованной таблицы позиций.
5. Нормализация массива позиций корзины
Сообщение о состоянии корзины содержит массив entries. Для аналитических операций каждая его позиция представляется отдельной строкой: это позволяет связывать позиции с каталогом, агрегировать количество и стоимость по товарным кодам, а также исследовать состав корзин.
Целевая таблица:
CREATE TABLE rabbitmq_queue_base.customer_cart_entry_data_extracted ( event_date DateTime DEFAULT now(), localpk_json Int64, localpk Nullable(Decimal(20, 0)), createdts DateTime, cartlocalpk Nullable(Decimal(20, 0)), modifiedts Nullable(DateTime), productcode Nullable(String), baseprice Nullable(Decimal(30, 8)), quantity Nullable(Decimal(30, 8)) ) ENGINE = ReplicatedMergeTree(…) PARTITION BY toYYYYMM(event_date) ORDER BY event_date TTL event_date + toIntervalDay(90);
Материализованное представление последовательно извлекает массив, разворачивает его и интерпретирует каждый элемент.
CREATE MATERIALIZED VIEW rabbitmq_queue_base.customer_cart_entry_mv_ext_json TO rabbitmq_queue_base.customer_cart_entry_data_extracted AS WITH customer_cart_data_array AS ( SELECT event_date, JSONExtractArrayRaw(raw_message, ‘entries’) AS entries_str, JSONExtractInt(raw_message, ‘localPk’) AS localpk FROM rabbitmq_queue_base.customer_cart_data ), customer_cart_data_flatten AS ( SELECT event_date, localpk, entries_str FROM customer_cart_data_array LEFT ARRAY JOIN entries_str ) SELECT event_date, localpk AS localpk_json, parsed.1 AS localpk, parseDateTimeBestEffort(parsed.2) AS createdts, parsed.4 AS cartlocalpk, parseDateTimeBestEffortOrNull(parsed.3) AS modifiedts, parsed.5 AS productcode, parsed.6 AS baseprice, parsed.7 AS quantity FROM ( SELECT event_date, localpk, JSONExtract( entries_str, ‘Tuple( localPk Nullable(Decimal(20, 0)), createdTs String, modifiedTs Nullable(String), cartLocal Nullable(Decimal(20, 0)), productCode Nullable(String), basePrice Nullable(Decimal(30, 8)), quantity Nullable(Decimal(30, 8)) )’ ) AS parsed FROM customer_cart_data_flatten WHERE entries_str != » ) AS raw_entry;
Последовательность преобразования следующая:
-
JSONExtractArrayRaw извлекает entries как массив исходных JSON-строк, а JSONExtractInt извлекает идентификатор корзины localPk.
-
LEFT ARRAY JOIN преобразует одну строку с массивом в набор строк — по одной на каждый элемент entries.
-
JSONExtract(…, ‘Tuple(…)’) применяется к отдельному элементу массива и извлекает его атрибуты.
-
Идентификатор родительской корзины передаётся в localpk_json. Он используется для связи строки позиции с заголовком корзины без повторного разбора исходного сообщения.
Источником представления является customer_cart_data, то есть структурирование заголовка корзины и нормализация её позиций выполняются на одном слое исходных сообщений.
Взаимосвязь объектов потока корзины:
customer_cart_queue │ ▼ customer_cart_mv ──────────────────────────► customer_cart_data │ ┌──────────────────────────┼──────────────────────────┐ ▼ ▼ │ customer_cart_mv_ext_json customer_cart_entry_mv_ext_json │ │ │ │ ▼ ▼ │ customer_cart_data_extracted customer_cart_entry_data_extracted │
6. Последовательность развёртывания
Объекты должны создаваться в порядке, соответствующем зависимостям. Для одного потока последовательность имеет вид:
-
*_data
-
*_data_extracted (и *_entry_data_extracted, если есть)
-
*_queue
-
*_mv
-
*_mv_ext_json (и *_entry_mv_ext_json)
На момент создания Materialized View целевая таблица, указанная в TO, уже должна существовать. В противном случае команда CREATE MATERIALIZED VIEW завершится ошибкой.
После развёртывания публикация сообщения в RabbitMQ инициирует его последовательное прохождение через таблицу-консьюмер, слой исходных данных и представления, формирующие структурированные таблицы.
7. Примеры аналитических запросов
Количество событий веб-трекинга по дням:
SELECT toDate(event_date) AS day, count() AS events FROM rabbitmq_queue_base.customer_tracking_data_extracted WHERE event_date >= today() — 7 GROUP BY day ORDER BY day;
Поисковые запросы, для которых не зафиксирован результат:
SELECT searchtext, count() AS cnt FROM rabbitmq_queue_base.customer_search_tracking_data_extracted WHERE event_date >= now() — INTERVAL 1 DAY AND (resultscount = 0 OR resultscount IS NULL) AND searchtext IS NOT NULL GROUP BY searchtext ORDER BY cnt DESC LIMIT 50;
Товарные коды с наибольшим числом позиций в корзинах:
SELECT productcode, count() AS lines, sum(quantity) AS qty FROM rabbitmq_queue_base.customer_cart_entry_data_extracted WHERE event_date >= today() — 7 AND productcode IS NOT NULL GROUP BY productcode ORDER BY lines DESC LIMIT 100;
Связь реквизитов корзины с её позициями:
SELECT c.code, c.user, c.totalprice, e.productcode, e.quantity, e.baseprice FROM rabbitmq_queue_base.customer_cart_data_extracted AS c INNER JOIN rabbitmq_queue_base.customer_cart_entry_data_extracted AS e ON c.localpk = e.localpk_json AND c.event_date = e.event_date WHERE c.event_date >= today() — 1 LIMIT 100;
Состав объектов
Объект
Тип
Назначение
*_queue
Table, RabbitMQ
Получение сообщения брокера в raw_message
*_mv
Materialized View
Запись исходного сообщения в таблицу *_data
*_data
Table, ReplicatedMergeTree
Хранение исходного JSON и времени загрузки
*_mv_ext_json
Materialized View
Преобразование JSON в типизированный набор атрибутов
*_data_extracted
Table, ReplicatedMergeTree
Хранение структурированных событий
customer_cart_entry_mv_ext_json
Materialized View
Разворачивание и разбор массива позиций корзины
customer_cart_entry_data_extracted
Table, ReplicatedMergeTree
Хранение нормализованных позиций корзины
Выводы
Рассмотренный конвейер разделяет приём сообщений, хранение исходного JSON и формирование структурированного слоя. В результате аналитики получают как исходные логи для диагностики и повторной обработки, так и типизированные таблицы для регулярных запросов.
Существенным элементом реализации является извлечение набора атрибутов одним вызовом JSONExtract с типом Tuple(…). После получения кортежа значения передаются в целевые колонки через обращения parsed.N, без повторного разбора исходной JSON-строки для каждого поля. Такой подход сокращает число операций извлечения JSON и явных преобразований типов в Materialized View.
В совокупности с пакетной обработкой сообщений, Materialized View и таблицами ReplicatedMergeTree использование Tuple позволило обеспечить на стенде обработку нагрузки до 1 000 сообщений в секунду; все сообщения становились доступны в ClickHouse менее чем за семь секунд. В сопоставимом нагрузочном тесте с альтернативной реализацией разбора JSON максимальная производительность составила 700 сообщений в секунду. Следовательно, применение Tuple повысило измеренный предел обработки на 300 сообщений в секунду, или приблизительно на 43%, и стало значимым фактором достижения требуемой пропускной способности.
Загрузка...
