Организация конвейера загрузки событий из RabbitMQ в ClickHouse

Постановка задачи

Задача была немного необычной для нашего DWH на основе Arenadata DB / Greenplum. А именно в DWH требовалось организовать приём событий пользовательского поведения из RabbitMQ и дополнительно предоставить команде e-commerce доступ к данным с задержкой не более пяти секунд. Под задержкой далее понимается интервал между публикацией сообщения в RabbitMQ и доступностью соответствующей записи для запроса в СУБД. Была выбран следующая схема обработки данных - потоковая запись из RabbitMQ в ClickHouse и минибатчинг из ClickHouse в Arenadata DB. В статье рассматривается первая часть схемы - интеграция RabbitMQ и ClickHouse. В самой статье рассматриваются несколько хинтов, которые помогли ускорить обработку данных и корректно обрабатывать сами сообщения (особенно это касается крзин).

В исходных именах таблиц и представлений используется термин customer tracking. В данной статье он обозначает события взаимодействия пользователя с платформой интернет-магазина: просмотры и переходы, поисковые запросы, операции с корзиной и действия в мобильном приложении. То есть термин clickstream events.

Сразу было принято решение записывать в отдельные таблицы:

1. таблица с исходными сообщения в виде JSON-логов — для диагностики, аудита преобразований и повторного разбора;

2. таблица с типизированными колонками — для аналитических запросов без регулярного парсинга 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;

При поддержке такого представления необходимо учитывать три аспекта:

1. Извлечение выполняется позиционно. Соответствие parsed.N полю сообщения определяется декларацией типа Tuple(...) и должно быть зафиксировано в контракте источника. Порядок колонок целевой таблицы не является основанием для такого соответствия.
   Здесь нужно отметить, что именно позиционное извлечение позволило достичь целевых показателей при нагрузочном тестировании.

2. Строковые представления дат преобразуются функциями parseDateTimeBestEffort и parseDateTimeBestEffortOrNull. Второй вариант применяется, когда отсутствие или некорректное значение даты считается допустимым.

3. Логические значения из 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;

Последовательность преобразования следующая:

1. JSONExtractArrayRaw извлекает entries как массив исходных JSON-строк, а JSONExtractInt извлекает идентификатор корзины localPk.

2. LEFT ARRAY JOIN преобразует одну строку с массивом в набор строк — по одной на каждый элемент entries.

3. JSONExtract(..., 'Tuple(...)') применяется к отдельному элементу массива и извлекает его атрибуты.

4. Идентификатор родительской корзины передаётся в 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. Последовательность развёртывания

Объекты должны создаваться в порядке, соответствующем зависимостям. Для одного потока последовательность имеет вид:

1. *_data

2. *_data_extracted (и *_entry_data_extracted, если есть)

3. *_queue

4. *_mv

5. *_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%, и стало значимым фактором достижения требуемой пропускной способности.