merges.txt

Оптимизация стратегии слияний.


Включить part_log. Для этого, раскомментируйте part_log в конфигурационном файле сервера.


Посмотреть на то, что происходит при вставке синтетических данных.

В clickhouse-client:

CREATE TABLE numbers (x UInt64) ENGINE = MergeTree ORDER BY (x);
INSERT INTO numbers SELECT * FROM system.numbers;

В clickhouse-client в другом терминале:

SELECT * FROM system.merges;
SELECT * FROM system.parts WHERE active AND table = 'numbers';
SELECT * FROM system.part_log;

Потом останавливаем запрос с помощью Ctrl+C.


Анализ будем производить на данных part_log.
Выгрузим part_log в tab separated файл:

clickhouse-client --query="SELECT * FROM system.part_log ORDER BY event_time" > part_log.tsv

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

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


Количественный анализ.

На основе part_log считаем следующие метрики:
- write amplification: отношение того, сколько данных было записаны изначально и в процессе мержей, к изначальному количеству данных;
- количество активных кусков в момент времени (квантили распределения: медиана, 90%, максимум);
- количество операций слияния в момент времени (квантили распределения: медиана, 90%, максимум);
- количество кусков в операции слияния (среднее);


Качественный анализ.

На основе part_log строим картинки:
- дерево слияний - как оно выглядит;
- график количества кусков по времени;
- график количества одновременных слияний по времени;


Симуляция алгоритма слияния.

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

Добавляем в эту программу имеющийся алгоритм из ClickHouse, чтобы можно было его менять и смотреть, на что влияют изменения.


Алгоритм выбора кусков для слияния в ClickHouse: SimpleMergeSelector.h
Описание PartLog: https://clickhouse.yandex/docs/ru/operations/server_settings/settings/#part_log а также PartLog.h


PS. Как я только что посчитал write amplification на основе part_log на живом сервере (для таблиц hits, visits):

example.yandex.ru :) WITH toUInt32(extract(part_name, '\\d+$')) AS level, level = 0 AND event_type IN ('NEW_PART', 'DOWNLOAD_PART') AS is_new, level > 0 AND event_type IN ('MERGE_PARTS', 'DOWNLOAD_PART') AS is_merged SELECT sumIf(rows, is_new) AS rows_new, formatReadableSize(sumIf(size_in_bytes, is_new) AS bytes_new) AS bytes_new_txt, sum(rows) AS rows_amplified, formatReadableSize(sum(size_in_bytes) AS bytes_amplified) AS bytes_amplified_txt, rows_amplified / rows_new AS rows_amplification, bytes_amplified / bytes_new AS bytes_amplification FROM system.part_log WHERE database = 'merge' AND table = 'visits' AND event_date >= today() - 30 AND NOT error

WITH 
    toUInt32(extract(part_name, '\\d+$')) AS level, 
    (level = 0) AND (event_type IN ('NEW_PART', 'DOWNLOAD_PART')) AS is_new, 
    (level > 0) AND (event_type IN ('MERGE_PARTS', 'DOWNLOAD_PART')) AS is_merged
SELECT 
    sumIf(rows, is_new) AS rows_new, 
    formatReadableSize(sumIf(size_in_bytes, is_new) AS bytes_new) AS bytes_new_txt, 
    sum(rows) AS rows_amplified, 
    formatReadableSize(sum(size_in_bytes) AS bytes_amplified) AS bytes_amplified_txt, 
    rows_amplified / rows_new AS rows_amplification, 
    bytes_amplified / bytes_new AS bytes_amplification
FROM system.part_log 
WHERE (database = 'merge') AND (table = 'visits') AND (event_date >= (today() - 30)) AND NOT error

┌───rows_new─┬─bytes_new_txt─┬─rows_amplified─┬─bytes_amplified_txt─┬─rows_amplification─┬─bytes_amplification─┐
│ 3844280818 │ 1.30 TiB      │    22780280604 │ 6.79 TiB            │  5.925758726401137 │   5.221187857249945 │
└────────────┴───────────────┴────────────────┴─────────────────────┴────────────────────┴─────────────────────┘

1 rows in set. Elapsed: 0.125 sec. Processed 2.72 million rows, 270.69 MB (21.66 million rows/s., 2.16 GB/s.) 

example.yandex.ru :) WITH toUInt32(extract(part_name, '\\d+$')) AS level, level = 0 AND event_type IN ('NEW_PART', 'DOWNLOAD_PART') AS is_new, level > 0 AND event_type IN ('MERGE_PARTS', 'DOWNLOAD_PART') AS is_merged SELECT sumIf(rows, is_new) AS rows_new, formatReadableSize(sumIf(size_in_bytes, is_new) AS bytes_new) AS bytes_new_txt, sum(rows) AS rows_amplified, formatReadableSize(sum(size_in_bytes) AS bytes_amplified) AS bytes_amplified_txt, rows_amplified / rows_new AS rows_amplification, bytes_amplified / bytes_new AS bytes_amplification FROM system.part_log WHERE database = 'merge' AND table = 'hits' AND event_date >= today() - 30 AND NOT error

WITH 
    toUInt32(extract(part_name, '\\d+$')) AS level, 
    (level = 0) AND (event_type IN ('NEW_PART', 'DOWNLOAD_PART')) AS is_new, 
    (level > 0) AND (event_type IN ('MERGE_PARTS', 'DOWNLOAD_PART')) AS is_merged
SELECT 
    sumIf(rows, is_new) AS rows_new, 
    formatReadableSize(sumIf(size_in_bytes, is_new) AS bytes_new) AS bytes_new_txt, 
    sum(rows) AS rows_amplified, 
    formatReadableSize(sum(size_in_bytes) AS bytes_amplified) AS bytes_amplified_txt, 
    rows_amplified / rows_new AS rows_amplification, 
    bytes_amplified / bytes_new AS bytes_amplification
FROM system.part_log 
WHERE (database = 'merge') AND (table = 'hits') AND (event_date >= (today() - 30)) AND NOT error

┌───rows_new─┬─bytes_new_txt─┬─rows_amplified─┬─bytes_amplified_txt─┬─rows_amplification─┬─bytes_amplification─┐
│ 3696083890 │ 712.31 GiB    │    43904455592 │ 6.22 TiB            │ 11.878641529426973 │   8.943485269922183 │
└────────────┴───────────────┴────────────────┴─────────────────────┴────────────────────┴─────────────────────┘

1 rows in set. Elapsed: 0.121 sec. Processed 2.72 million rows, 270.69 MB (22.50 million rows/s., 2.24 GB/s.)