OLAP OVER HTTP: как отдавать большие аналитические данные через API и не положить сервис

Каждый день в 18:00 клиентам становится доступна отчётность по продажам, и многие одновременно нажимают кнопку «Загрузить отчёт». Это может быть детализированный отчёт по списаниям, юридическая отчётность, аналитическая выгрузка или API, используя который клиент получает данные для дальнейшей обработки. В этот момент даже хорошо подобранное хранилище может стать узким местом.

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

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

• На запрашиваемые данные может быть большой RPS.

• Каждый запрос может отдавать довольно большой объём данных.

• Возникает «проблема звёзд».

Почему нельзя просто поставить endpoint над OLAP-базой

Решать эту проблему можно по-разному, но начинать всегда стоит с выбора наиболее подходящего хранилища данных. OLTP СУБД по типу PostgreSQL, MS SQL или MySQL не всегда подходят для сценариев потоковой отдачи аналитических данных, когда из множества атрибутов для отчёта требуется лишь ограниченная выборка. Это связано с вычислительной моделью, которая в них заложена: кортежи читаются один за другим, и каждый из них нужно проверить на MVCC-видимость, что создаёт большие накладные расходы (overhead). В то же время OLAP-хранилища используют другую вычислительную модель, без таких тяжёлых проверок — читают с диска большими батчами и выделяют на обработку запроса десятки потоков. В качестве OLAP-хранилища выбор часто падает на ClickHouse, но каждый волен выбрать то, что ему подходит больше или что уже используется в компании.

Ну вот, казалось, и ответ: чтобы предоставлять доступ к аналитическим данным по HTTP, надо просто подобрать аналитическую СУБД и сделать endpoint к ней. И в самом простом случае этого действительно будет достаточно, но у такого решения есть один подводный камень: OLAP-базы данных не держат большой RPS. Это связано с тем, что каждый запрос стремится использовать максимальное число ядер для параллельной обработки, что приводит к Thread Contention, Noisy Neighbor Problem и др. Такая проблема не обязательно связана с тем, что у вас всегда высокие нагрузки. Она может быть вызвана специфичными сценариями доступа к данным, например:

• Сезонными или событийными паттернами. Например, в Ozon нагрузки возрастают в 2–3 раза в сезоны распродаж.

• «Проблемой звёзд» в социальных сетях, когда один объект намного популярнее других и из-за этого все запросы к БД приходятся на одну партицию.

• Бесконечной прокруткой с дырами. Пользователь листает ленту глубоко вниз и, чтобы её обновлять, продолжает делать запросы в БД с большим OFFSET. Если запрос составлен некорректно или плохо настроены индексы, то база данных может начать кешировать результаты выполнения запроса на диске (external merge в плане запроса).

Экспериментальный стенд

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

В качестве OLAP-базы для тестов выбран ClickHouse. Все исходники можно найти на GitHub. В README описаны инструкции для запуска и повторения результатов. Для понимания сути статьи переходить в репозиторий необязательно.

Данные и схема таблицы

Для постановки проблемы сначала посмотрим, какую нагрузку выдержит чистый ClickHouse на 4 ГБ ОЗУ и 6 ядер ЦП с частотой 3,8 ГГц. Поднимем контейнер с базой и создадим следующую таблицу.

CREATE TABLE IF NOT EXISTS postings

(

    posting_id              Int64,

    posting_name            String,

    seller_id               Int64,

    item_ids                Array(Int64),

    item_quantities         Array(Int32),

    marketplace_item_prices Array(Decimal(18, 6)),

    seller_item_prices      Array(Decimal(18, 6)),

    seller_currency         Int8,

    seller_fx_rate          Decimal(18, 8),

    posting_created_at      DateTime,

    posting_delivered_at    DateTime,

    posting_source          LowCardinality(String),

    total_amount            Decimal(18, 6),

    payment_method          LowCardinality(String),

    shipping_country        LowCardinality(String),

    shipping_city           String,

    version                 Int8

)

ENGINE = ReplacingMergeTree(version)

PARTITION BY toYYYYMM(posting_created_at)

ORDER BY (seller_id, posting_created_at, posting_id);

Примечание: в ClickHouse можно указать тип сжатия колонок. В данном случае он не указан и используется дефолтный алгоритм LZ4. Если хочется поэкспериментировать со сжатием, например сильнее пожать данные в обмен на небольшое снижение производительности на чтение, то у колонки можно явно задать кодек ключевым словом CODEC(your_codec).

Сколько данных загружаем

Зальём в неё данные. В репозитории уже есть готовая утилита для этого. Я сгенерировал список из 5 000 случайных ID селлеров — он участвует в генерации данных и понадобится для выполнения запросов.

Сами данные я наполнял равномерно: всего 4 миллиона записей в месяц. За год набралось 48 миллионов.

Нагрузочное тестирование системы с ClickHouse

После заливки данных остаётся только сделать endpoint для доступа к ним и начать нагрузочное тестирование. Утилита для нагрузочного тестирования также есть в репозитории. Ниже привожу запрос, который будет выполняться.

WITH filtered_rows AS (

    SELECT

        posting_id,

        posting_name,

        posting_created_at,

        posting_delivered_at,

        item_ids,

        item_quantities,

        marketplace_item_prices,

        posting_source,

        payment_method,

        shipping_city,

        shipping_country

    FROM posting.postings

    WHERE seller_id = @sellerId

        AND posting_created_at BETWEEN @periodStart AND @periodEnd)

SELECT

    posting_id,

    any(posting_name),

    any(posting_created_at),

    any(posting_delivered_at),

    item_id,

    SUM(item_quantity),

    SUM(item_quantity * marketplace_item_price),

    posting_source,

    payment_method,

    shipping_city,

    shipping_country

FROM filtered_rows

ARRAY JOIN

    item_ids as item_id,

    item_quantities as item_quantity,

    marketplace_item_prices as marketplace_item_price

GROUP BY posting_id, item_id, posting_source, payment_method, shipping_city, shipping_country

Запрос довольно прямолинейный. У нас на практике бывали случаи и поинтереснее. Например, надо было загружать в ClickHouse временную таблицу и джойнить её к результату запроса. Был случай, когда из-за бизнес-требований надо было поддержать сортировку по двум датам в зависимости от параметров запросов, что также является задачей нетривиальной. Например, в ClickHouse таблицы физически хранятся в отсортированном виде, поэтому сортировать одну таблицу по двум датам не получится. Возможно, это тема для отдельной статьи.

Что и как тестируем

Сам сервис (ASP.NET Core) запущен на localhost на машине с 6 ядрами CPU с частотой 3,8 ГГц, NVMe SSD, все тестируемые СУБД подняты в Docker-контейнерах. Нагрузку генерирует отдельный консольный проект OlapOverHttp.LoadTest — он параллельно шлёт HTTP-запросы на endpoint сервиса, считает общее количество запросов и среднее время ответа.

Непосредственно в этом нагрузочном тесте мы проверяем два endpoint'а:

• GET /api/postings — запрос уходит в ClickHouse, который хранит данные за всё время;

• GET /api/postings/hot-cold — запрос идёт в PostgreSQL, где лежат только горячие данные.

Оба endpoint'а принимают одинаковые входные параметры — диапазон дат и идентификатор селлера, — чтобы обеспечить сопоставимость результатов. С помощью проекта LoadTest мы генерируем сначала нагрузку на первый ednpoint, затем на второй. В README есть инструкция по использованию этого проекта.

P. S. В реальном приложении endpoint hot-cold должен выбирать, в какую СУБД сходить за результатом в зависимости от параметров запроса. В целях проведения нагрузочного тестирования он ходит только в PostgreSQL.

Ремарка к результатам нагрузочного тестирования

В данном случае любой тестируемый компонент (база или BLOB-хранилище, про которое написано далее) запущен на localhost, как и сервис, который запрашивает из него данные. В самом же компоненте находится всего одна таблица или бакет, данные там залиты за год, в то время как в компоненте из продакшен-среды, скорее всего, будет несколько таблиц и данные за несколько лет, поэтому результаты тестирования в продакшене будут гораздо ниже.

Результаты теста

По результатам нагрузочного тестирования получаем, что ClickHouse держит примерно 60 RPS. Среднее время ответа ~80 ms. На 70 RPS начинается серьёзная деградация времени ответа, а на 80 RPS база полностью захлёбывается. Что же делать, если надо держать хотя бы 100 RPS?

Подход 1: оптимизация уровня хранения — стратегия hot/cold storage

Одним из возможных решений является применение стратегии hot/cold storage. Она исходит из предпосылки о том, что можно выделить данные, к которым пользователи обращаются чаще всего. Например, им наиболее интересны данные за последние полгода. И на самом деле эта предпосылка выполняется очень часто. Согласно нашим дашбордам, 99% запросов приходятся именно на горячие данные. Рассмотрим реализацию этой стратегии по шагам:

1. Выбираем и создаём СУБД, которая хорошо держит RPS. Зачастую подойдёт и типичное OLTP-решение по типу PostgreSQL.

2. Воссоздаём в ней схему из долгосрочного хранилища.

3. Начинаем записывать поступающие данные в новую БД. Для этого, возможно, придётся изменить абстракцию доступа к данным.

4. Переливаем данные за выбранный период из старой БД в новую.

5. В endpoint'е контроллера добавляем развилку для выбора БД (if (CanQueryHotStorage(request.Date))).

6. Добавляем джобу для очищения неактуальных данных из «горячего» хранилища данных.

Hot storage на PostgreSQL

Для развития нашего примера требуется только выполнить шаги 1, 2, 4 и 5. В качестве OLTP-базы я выбрал PostgreSQL. Миграцию и запрос предлагаю посмотреть в листинге ниже. В нашем случае перенос схемы — задача не самая сложная, но в реальности подбор оптимальной структуры таблиц, индексов и их тюнинг может занять существенное время.

CREATE SCHEMA IF NOT EXISTS "posting";

CREATE TABLE IF NOT EXISTS "posting".postings(
	id                   bigserial PRIMARY KEY,
	posting_id           bigint NOT NULL,
	posting_name         text STORAGE MAIN NOT NULL,
	seller_id            bigint NOT NULL,
	seller_fx_rate       decimal(18, 6) NOT NULL,
	seller_currency      int8 NOT NULL,
	posting_created_at   timestamptz NOT NULL,
	posting_delivered_at timestamptz NOT NULL,
	posting_source       text STORAGE MAIN NOT NULL,
	total_amount         decimal(18, 4) NOT NULL,
	payment_method       text STORAGE MAIN NOT NULL,
	shipping_country     text STORAGE MAIN NOT NULL,
	shipping_city        text STORAGE MAIN NOT NULL);

CREATE UNIQUE INDEX IF NOT EXISTS postings_posting_id_seller_id ON "posting".postings(posting_id, seller_id);

CREATE INDEX IF NOT EXISTS postings_seller_id_posting_created_at_idx ON "posting".postings(seller_id, posting_created_at);

CREATE TABLE IF NOT EXISTS "posting".items(
	posting_entry_id       bigint REFERENCES "posting".postings(id),
	item_id                bigint NOT NULL,
	version                int NOT NULL,
	item_quantity          int NOT NULL,
	marketplace_item_price decimal(18, 4) NOT NULL,
	seller_item_price      decimal(18, 4) NOT NULL,
	PRIMARY KEY (posting_entry_id, item_id)
);

WITH filtered_posting_with_items AS (
            SELECT
                p.posting_id,
                p.posting_created_at,
                p.posting_delivered_at,
                p.posting_name,
                p.seller_fx_rate,
                p.seller_currency,
                p.posting_source,
                p.payment_method,
                p.shipping_country,
                p.shipping_city,
                i.item_id,
                i.item_quantity,
                i.marketplace_item_price,
                i.seller_item_price
            FROM posting.postings p
            JOIN posting.items i ON p.id = i.posting_entry_id
            WHERE seller_id = @sellerId
                AND posting_created_at BETWEEN @periodStart AND @periodEnd 
        )
        SELECT
            any_value(posting_name) AS PostingName,
            posting_created_at AS PostingDate,
            any_value(posting_delivered_at) AS DeliveryDate,
            item_id AS ItemId,
            SUM(item_quantity)::int AS ItemQuantity,
            SUM(item_quantity * marketplace_item_price) AS ItemTotal,
            posting_source AS PostingSource,
            payment_method AS PaymentMethod,
            shipping_city AS ShippingCity,
            shipping_country AS ShippingCountry
        FROM filtered_posting_with_items
        GROUP BY posting_id, posting_created_at, item_id, posting_source, payment_method, shipping_country, shipping_city

Обратим внимание на схему и запрос. Предположим, что мы загружаем данные за последний месяц по тем же самым 5 000 селлерам, то есть в сумме у нас 4 миллиона записей. В случае равномерного распределения данных запрос по каждому селлеру будет затрагивать 0,02% всех строк и 800 строк в целом. В рассмотренном сценарии планировщик запросов будет всегда строить запросы только с использованием Index Scan. В итоге у нас получаются быстрые и небольшие запросы. На практике делать заливку можно за срок гораздо больший, чем один месяц, но это надо тестировать — распределение может быть неравномерным, данных может быть больше.

Утилита для выполнения пункта 4 также есть в решении, в проекте OlapOverHttp.Postgres.Filler. Я переливал данные из ClickHouse за последние 2 месяца. Думаю, что 8 миллионов строк достаточно, чтобы продемонстрировать получаемый эффект. Перейдём к результату теста.

Примечание по тестированию PostgreSQL: базу надо сначала разогреть. Для этого можно запустить нагрузочное тестирование несколько раз или на более длинный срок, чтобы база смогла загрузить нужные страницы в кеш.

PostgreSQL на нашей выборке и железе держит примерно 240 RPS. На 250 RPS уже начинаются ошибки too many clients, но это в 4 раза больше, чем у ClickHouse. Что ещё более интересно, так это то, что среднее время ответа составило 12 ms, что в 6,7 раза меньше, чем у холодного хранилища.

Тут главное — не подумать, что PostgreSQL в целом лучше, чем ClickHouse, и что изначально надо было делать всё на нём. Не стоит забывать, что ClickHouse создан для быстрых аналитических запросов на сотни миллионов строк, которые PostgreSQL едва ли сможет адекватно переварить.

Итак, в результате мы получаем систему, в которой каждый компонент занимается тем, что у него получается лучше всего: горячее хранилище быстро отдаёт данные за короткий промежуток времени, холодное хранилище хранит все архивные данные и периодически предоставляет доступ к ним, если пользователю это вдруг потребуется. Очень часто ещё выходит значительно снизить время отклика, что на моём стенде также подтвердилось. Из интересных добавлений могу отметить, что данные в OLTP-хранилище можно записывать сразу в агрегированном виде, чтобы не выполнять группировку во время запроса. Это получится сделать далеко не всегда, но даст ещё больший прирост к скорости работы.

Подход 2: оптимизация вычислений — кеширование результатов запросов

Второй возможной оптимизацией, которую я хотел рассмотреть, является кеширование результатов запросов. Тут тоже есть предпосылка, из которой мы исходим: кеширование должно иметь смысл. Например, запрос, который мы выполняем, имеет строгие временные рамки и направлен на конкретного клиента или другую фиксированную сущность. В Ozon у нас есть подобный кейс: нам часто требуется сгенерировать отчёт с детализацией списываний средств по какой-то услуге. Такой отчёт формируется только после определённой даты, за фиксированный период и направлен на конкретного селлера, а значит, хорошо кешируется. Результаты предоставляются селлерам в виде Excel-файла, поэтому в примере на стенде я тоже буду генерировать .xlsx файлы.

Из важных уточнений хочется отметить, что такой набор условий позволяет генерировать отчёты заранее, чтобы когда селлеры пойдут скачивать отчёты, они уже были в кеше. Такая стратегия кеширования позволяет иметь 100% cache hit, и именно её я реализовал в своём решении.

Реализация этого решения довольно прямолинейная: выбираем кеш, добавляем его в проект и начинаем записывать и читать отчёты из кеша. В случае генерации файлов в качестве кеша может выступать S3-совместимое объектное хранилище. В своём проекте я использовал MinIO. Тут важно не забыть учесть два подводных камня:

1. Что делать в случае, если в отчёте выявилась ошибка и его требуется пересоздать, и могут ли быть такие случаи?

2. Что делать, если в кеше не нашлось нужного отчёта? Возвращать 404 или генерировать его на лету?

В моём проекте я выбрал MinIO, поскольку его оказалось проще встроить в тестовый стенд. В проде вместо него может использоваться другое S3-совместимое объектное хранилище. Слой доступа к отчётам выглядит так:

public async Task GenerateCachedReport(

    ReportRequest request,

    PipeWriter writer,

    CancellationToken token)

{

    // Получаем путь, по которому должен храниться отчёт в кеше

    var reportObjectName = request.GetObjectName();

    // Пытаемся загрузить его из кеша. Если не получилось, генерируем отчёт и загружаем его в кеш и качаем оттуда

    var reportStream = await objectStorage.Download(reportObjectName)

        ?? await GenerateCachedReportAndDownload(request, token);

    // если по какой-то причине кеш отвалился, то просто качаем отчёт из ClickHouse

    if (reportStream is not null)

        await reportStream.CopyToAsync(writer);

    else

        await GenerateReport(request, writer, token);

}

В моей реализации я решил, что если отчёта нет в кеше, то он генерируется и загружается в него. Если кеш по какой-то причине недоступен, то выполняется генерация на лету. Полную версию можно посмотреть на GitHub.

Результаты теста с кешированием

Теперь проведём сравнительное нагрузочное тестирование и посмотрим результаты:

Результаты хоть и показывают огромную разницу в производительности, но являются довольно предсказуемыми. В первом случае мы тестируем, какой RPS держит база, во втором — кеш, но это не отменяет того, что прирост огромный. Сама генерация отчётов за месяц для 5 000 селлеров заняла 61 секунду. Обслуживание 5 000 запросов без кеширования займёт 5 000 / 70 = 71,42 секунды, а с кешированием 61 + 5 000 / 750 = 67,67 секунды (я не подгонял цифры). То есть даже на 5 000 запросах подход с кешированием выигрывает. Если нам приходит 50 000 запросов по тем же селлерам, то выгода становится более явная: 714,29 против 127,67 секунды. 

А что самое главное, мы опять получаем систему, в которой каждый компонент занимается тем, что у него получается хорошо: S3-хранилище держит RPS, а OLAP-база записывает в него результаты запроса. А устранение повторных запросов значительно снижает нагрузку на БД.

Рассмотрим ещё ряд улучшений, которые можно применить:

1. Применение политики сжатия данных. Сжатые данные занимают меньше места на диске, а также быстрее загружаются в память, но требуют больше ресурсов CPU. Важно, что для каждого конкретного workflow преимущества неочевидны и требуют экспериментального подтверждения на ваших данных.

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

3. Тюнинг БД. Да, это очевидный пункт, но я хочу его упомянуть, чтобы обратить внимание на то, что OLAP-БД настраивается не так, как OLTP, в силу очевидных различий в дизайне. Например, в том же ClickHouse крайне важно правильно подобрать поля в OrderBy.

4. Пул соединений. Если для PostgreSQL уже есть зарекомендовавшие себя решения для управления пулом соединений, то для ClickHouse этой задачей придётся озаботиться на уровне кода.

Конечно, этим список не ограничивается, и в зависимости от каждого конкретного сценария нагрузки можно найти что-то ещё.
В конце скажу: всегда проводите нагрузочное тестирование ваших решений перед выкаткой в прод. В моей практике нередко бывало, что решения, которые хорошо смотрелись в коде, имели недостатки, которые были заметны только при нагрузке. Например, выбранный кодек может давать неоптимальную степень сжатия на ваших данных, а распаковка — потреблять неожиданно много CPU. Планировщик запросов может посчитать использование индекса неоптимальным на реальных запросах (у нас такое было на запросах с большим offset). Кроме того, конфигурация железа (скорость диска, количество ядер, пропускная способность памяти) сильно влияет на то, где именно возникает узкое место. Выявление таких ошибок до ввода в эксплуатацию спасёт вам кучу нервных клеток.

Вывод

ClickHouse и другие OLAP-хранилища хорошо подходят для тяжёлых аналитических запросов, но это не значит, что они должны напрямую обслуживать каждый пользовательский HTTP-запрос. Частые запросы к актуальным данным можно вынести в hot storage, регулярно генерирующиеся отчёты — кешировать в объектном хранилище, а тяжёлые операции — выполнять заранее или в периоды низкой активности.

Так мы получаем систему, где каждый компонент занимается своей задачей: ClickHouse хранит и обрабатывает большие объёмы данных, hot storage быстро отвечает на частые запросы, S3 отдаёт готовые файлы, а API не превращается в тонкую прокладку над тяжёлой базой.

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