Ускорение аналитики с MergeTree

Загружайте данные из открытых табличных форматов в таблицы ClickHouse MergeTree, чтобы значительно ускорить аналитические запросы.

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

У MergeTree есть несколько преимуществ по сравнению с прямым чтением открытых табличных форматов:

• разреженный первичный индекс - Упорядочивает данные на диске по выбранному ключу, благодаря чему ClickHouse может пропускать большие диапазоны нерелевантных строк при выполнении запросов.
• Расширенные типы данных - Встроенная поддержка таких типов, как JSON, LowCardinality и Enum, обеспечивает более компактное хранение и более быструю обработку.
• Индексы пропуска и полнотекстовые индексы - Вторичные индексные структуры, которые позволяют ClickHouse пропускать гранулы, не соответствующие предикатам фильтра в запросе; это особенно эффективно для рабочих нагрузок текстового поиска.
• Быстрые вставки с автоматической компакцией - ClickHouse рассчитан на вставки с высокой пропускной способностью и автоматически выполняет слияние частей данных в фоновом режиме, аналогично compaction в открытых табличных форматах.
• Оптимизировано для параллельного чтения - Столбцовая структура хранения MergeTree в сочетании с несколькими уровнями кэширования поддерживает аналитические рабочие нагрузки в реальном времени с высоким параллелизмом — то, для чего открытые табличные форматы не предназначены.

В этом руководстве показано, как загружать данные из каталога в таблицу MergeTree с помощью INSERT INTO SELECT, чтобы ускорить аналитику.

​

Подключение к каталогу

Мы будем использовать то же подключение к Unity Catalog из предыдущего руководства и подключаться через конечную точку Iceberg REST:

SET allow_database_iceberg = 1;

CREATE DATABASE unity
ENGINE = DataLakeCatalog('https://<workspace-id>.cloud.databricks.com/api/2.1/unity-catalog/iceberg-rest')
SETTINGS catalog_type = 'rest', catalog_credential = '<client-id>:<client-secret>', warehouse = 'workspace',
oauth_server_uri = 'https://<workspace-id>.cloud.databricks.com/oidc/v1/token', auth_scope = 'all-apis,sql';

​

Список таблиц

SHOW TABLES FROM unity

┌─name───────────────────────────────────────────────┐
│ unity.logs                                         │
│ unity.single_day_log                               │
└────────────────────────────────────────────────────┘

​

Изучите схему

SHOW CREATE TABLE unity.`icebench.single_day_log`

CREATE TABLE unity.`icebench.single_day_log`
(
    `pull_request_number` Nullable(Int64),
    `commit_sha` Nullable(String),
    `check_start_time` Nullable(DateTime64(6, 'UTC')),
    `check_name` Nullable(String),
    `instance_type` Nullable(String),
    `instance_id` Nullable(String),
    `event_date` Nullable(Date32),
    `event_time` Nullable(DateTime64(6, 'UTC')),
    `event_time_microseconds` Nullable(DateTime64(6, 'UTC')),
    `thread_name` Nullable(String),
    `thread_id` Nullable(Decimal(20, 0)),
    `level` Nullable(String),
    `query_id` Nullable(String),
    `logger_name` Nullable(String),
    `message` Nullable(String),
    `revision` Nullable(Int64),
    `source_file` Nullable(String),
    `source_line` Nullable(Decimal(20, 0)),
    `message_format_string` Nullable(String)
)
ENGINE = Iceberg('s3://...')

Эта таблица содержит ~283 миллиона строк логов из тестовых запусков ClickHouse CI — реалистичный набор данных для оценки производительности аналитических запросов.

SELECT count()
FROM unity.`icebench.single_day_log`

┌───count()─┐
│ 282634391 │ -- 282,63 миллиона
└───────────┘

1 строка в наборе. Elapsed: 1.265 sec.

​

Запрос к таблице озера данных

Выполним запрос, который фильтрует журналы по имени потока и типу инстанса, ищет в тексте сообщения ошибки и группирует результаты по логгеру:

SELECT
    logger_name,
    count() AS c
FROM icebench.`icebench.single_day_log`
WHERE (thread_name = 'TCPHandler')
    AND (instance_type = 'm6i.4xlarge')
    AND hasToken(message, 'error')
GROUP BY logger_name
ORDER BY c DESC
LIMIT 5

┌─logger_name──────────────┬────c─┐
│ executeQuery             │ 6907 │
│ TCPHandler               │ 4145 │
│ TCP-Session              │  790 │
│ PostgreSQLConnectionPool │  530 │
│ ContextAccess (default)  │  392 │
└──────────────────────────┴──────┘

5 rows in set. Elapsed: 8.921 sec. Processed 282.63 million rows, 5.42 GB (31.68 million rows/s., 607.26 MB/s.)
Пиковое потребление памяти: 4.35 GiB.

Запрос выполняется почти 9 секунд, потому что ClickHouse приходится полностью сканировать таблицу по всем файлам Parquet в объектном хранилище. Производительность можно было бы повысить с помощью партиционирования, но у столбцов вроде logger_name может быть слишком высокая мощность, чтобы партиционирование было эффективным. У нас также нет индексов, таких как Текстовые индексы, чтобы ещё сильнее отсекать данные. Именно здесь MergeTree показывает себя лучше всего.

​

Загрузите данные в MergeTree

​

Создание оптимизированной таблицы

Создадим таблицу MergeTree, немного оптимизировав схему. Обратите внимание на несколько ключевых отличий от схемы Iceberg:

• Без обёрток Nullable — отказ от Nullable повышает эффективность хранения и производительность запросов.
• LowCardinality(String) для столбцов level, instance_type, thread_name и check_name — использует словарное кодирование для столбцов с небольшим числом различных значений, что улучшает сжатие и ускоряет фильтрацию.
• полнотекстовый индекс для столбца message — ускоряет полнотекстовый поиск по токенам, например hasToken(message, 'error').
• Ключ ORDER BY (instance_type, thread_name, toStartOfMinute(event_time)) — организует данные на диске в соответствии с типичными условиями фильтрации, чтобы разреженный первичный индекс мог пропускать нерелевантные гранулы.

SET enable_full_text_index = 1;

CREATE TABLE single_day_log
(
    `pull_request_number` Int64,
    `commit_sha` String,
    `check_start_time` DateTime64(6, 'UTC'),
    `check_name` LowCardinality(String),
    `instance_type` LowCardinality(String),
    `instance_id` String,
    `event_date` Date32,
    `event_time` DateTime64(6, 'UTC'),
    `event_time_microseconds` DateTime64(6, 'UTC'),
    `thread_name` LowCardinality(String),
    `thread_id` Decimal(20, 0),
    `level` LowCardinality(String),
    `query_id` String,
    `logger_name` String,
    `message` String,
    `revision` Int64,
    `source_file` String,
    `source_line` Decimal(20, 0),
    `message_format_string` String,
    INDEX text_idx(message) TYPE text(tokenizer = splitByNonAlpha)
)
ENGINE = MergeTree
ORDER BY (instance_type, thread_name, toStartOfMinute(event_time))

​

Вставка данных из каталога

Используйте INSERT INTO SELECT, чтобы загрузить ~300 млн строк из таблицы озера данных в нашу таблицу ClickHouse:

INSERT INTO single_day_log SELECT * FROM icebench.`icebench.single_day_log`

282634391 rows in set. Elapsed: 237.680 sec. Processed 282.63 million rows, 5.42 GB (1.19 million rows/s., 22.79 MB/s.)
Peak memory usage: 18.62 GiB.

​

Выполните запрос повторно

Если теперь снова выполнить тот же запрос к таблице MergeTree, мы увидим, что производительность значительно возрастет:

SELECT
    logger_name,
    count() AS c
FROM single_day_log
WHERE (thread_name = 'TCPHandler')
    AND (instance_type = 'm6i.4xlarge')
    AND hasToken(message, 'error')
GROUP BY logger_name
ORDER BY c DESC
LIMIT 5

┌─logger_name──────────────┬────c─┐
│ executeQuery             │ 6907 │
│ TCPHandler               │ 4145 │
│ TCP-Session              │  790 │
│ PostgreSQLConnectionPool │  530 │
│ ContextAccess (default)  │  392 │
└──────────────────────────┴──────┘

5 rows in set. Elapsed: 0.220 sec. Processed 13.84 million rows, 2.85 GB (62.97 million rows/s., 12.94 GB/s.)
Peak memory usage: 1.12 GiB.

Теперь тот же запрос выполняется за 0,22 секунды — примерно в 40 раз быстрее. Это улучшение обеспечивают две ключевые оптимизации:

• Разреженный первичный индекс - ключ ORDER BY (instance_type, thread_name, ...) означает, что ClickHouse может сразу перейти к гранулам, соответствующим instance_type = 'm6i.4xlarge' и thread_name = 'TCPHandler', сократив число обрабатываемых строк с 283 миллионов до всего 14 миллионов.
• Полнотекстовый индекс - индекс text_idx для столбца message позволяет обрабатывать hasToken(message, 'error') по индексу, а не сканировать каждое сообщение целиком, что ещё больше уменьшает объём данных, которые ClickHouse нужно прочитать.

В результате получается запрос, который без проблем может обеспечивать работу панели мониторинга в реальном времени — в масштабе и с задержкой, недостижимыми для запросов к файлам Parquet в Объектном хранилище.