Оптимизация SELECT-ов с использованием словарей

[Stepan Semiokhin]

Добрый день!

У нас возник следующий кейс: есть 3 машины с распределенной таблицей (машины хорошие, по 128гб оперативки каждая), которая базируется на MergeTree-таблицах.

Один из запросов активно использует hashed-словрь (достаточно большого размера, 47к записей) имеет вид:

SELECT

concat(toString(col1_code), concat(' ', concat(col1, concat(' -> ', concat(toString(col2_code), ' END')))) as path, 

concat(toString(col1_code), concat(' ', col1)) as col1,

....

sum(c_0) > 0 ? sum(c_0) : -1 as c_0,

sum(c_1) > 0 ? sum(c_1) : -1 as c_1,

...

some_idx

FROM (

SELECT

toUInt8(dictGetString('dict', 'type', toUInt64(key_code))) = 1 ? key_code

: dictGetInt32('dict', 'col1_index', toUInt64(key_code)) 

as col1_code,

toUInt8(dictGetString('dict', 'type', toUInt64(key_code))) = 1 ? dictGetStringOrDefault('dict', 'object_border', toUInt64(key_code), dictGetString('dict', 'post_object', toUInt64(key_code))) 

: dictGetString('dict', 'col1', toUInt64(key_code))

as col1,

...

dictGetStringOrDefault('dict', 'object_border', toUInt64(key_code), dictGetString('dict', 'post_object', toUInt64(key_code))) as key,

key_code,

c_0, 

c_1, 

...

some_idx

FROM table

WHERE key_code_flag != 0 

)

WHERE notEmpty(key) = 1

GROUP BY some_idx, col1_code, col1, col2_code, col2, key_code, key

ORDER BY c_10 desc

Он большой, я представил обобщенно основные моменты, *_code - это столбцы типа Int32, остальные - это String.

На 270 млн. строк данный вопрос выполняется ~42-43 сек , что довольно много относительно HP Vertica (вместо которой и планировать использовать Кликхаус).

Скажите, пожалуйста, как можно оптимизировать данный запрос или структуру (например, не использовать словари, а просто JOIN-ить)? Связано ли это с множеством приведений типов?

[Stepan Semiokhin]

Вот, например, что меня смутило в логе:

MemoryTracker: Peak memory usage (for query): 857.15 MiB.

Хотя max_memory_usage - 10 гб, смотреть в направлении max_rows_to_read , max_rows_to_group_by, etc?

[Stepan Semiokhin]

Пока эксперименты с этими параметрами не дали никакого результата, те же 830Мб-1гб по оперативке

четверг, 21 июля 2016 г., 11:47:59 UTC+3 пользователь Stepan Semiokhin написал:

[man...@gmail.com]

1. Для функции concat уже поддерживается версия с произвольным количеством аргументов.
Надо добавить в документацию...

2. "concat(toString(col1_code), concat(' ', col1)) as col1"
Алиасы в ClickHouse видны глобально. Поэтому, когда вы указываете в GROUP BY col1, это будет не исходный col1 таблицы, а результат concat - довольно сложное выражение. Чтобы не было путаницы, можно использовать для алиаса другое имя.

3. На всякий случай - существует функция concatAssumeInjective. Она предназначена скорее не для людей, а для использования в генераторах запросов в стороннем коде. Эта функция имеет такой же смысл как concat, за исключением того, что она, в целях оптимизации запроса, считается взаимно-однозначной, и поэтому, при использовании в GROUP BY, заменяется на свои аргументы.

4. "toUInt8(dictGetString(...))"
Зачем так? Конечно, парсинг числа из строки - недешёвая операция, даже не смотря на то, что она хорошо оптимизирована.
Лучше избежать лишних преобразований между числами и строками.

5. "hashed-словрь (достаточно большого размера, 47к записей)"
Словари работают эффективнее чем JOIN. Они, по сути, представляют собой, специализированный вариант JOIN.
47 тысяч записей - небольшой размер для словаря.
Если все ключи меньше 500 000, то можно использовать "flat" размещение в памяти, оно чуть-чуть лучше.
(Но судя по порядку времени, разница в вашем случае будет небольшой.)

6. При первом выполнении запроса, будет тратиться время на загрузку словаря.

7. Если указано, что использовалось 857 MiB памяти, значит в большем объёме, для данного запроса, не было необходимости.
Также стоит иметь ввиду, что распределённый запрос порождает запросы на удалённых серверах. Для каждого из них выводится потребление памяти.
Потребление памяти на удалённых серверах и на сервере-инициаторе запроса будет разным.

8. "WHERE notEmpty(key) = 1"
Можно писать просто "WHERE notEmpty(key)".

четверг, 21 июля 2016 г., 12:14:22 UTC+3 пользователь Stepan Semiokhin написал:

[Stepan Semiokhin]

Спасибо большое, учту!

А по настройке не может подсказать? Просто мощности есть, 32 ядра виртуальных (16 физических), оперативки 190гб (а не 128, как мне казалось), единовременно планируются не так много запросов (10-15 в пике).

Я попробовал ряд стандартных настроек вроде увеличения max_block_size или max_threads/max_distributed_processing_threads, но это особо не сказалось на времени выполнения запроса.

Вообще на таблице в 267 млн. записей (по 89 млн. на каждой из 3-х машин) я получаю время обычного селекта по всем данным (select * from dist_table;) равное ~19-20 сек.

Нормальный ли это результат или при правильной настройке можно получить результат лучше?

 

четверг, 21 июля 2016 г., 16:56:30 UTC+3 пользователь man...@gmail.com написал:

[Stepan Semiokhin]

После указанных вами оптимизаций действительно удалось "съесть" 8 секунд, спасибо!

Выявил, что обращаясь к обычной MergeTree-таблице запрос выполняется 6 секунд, получается, основную часть времени выполняется merge и "доагрегация" на сервере-инициаторе... Могу ли я как-либо повлиять на этот процесс?

[man...@gmail.com]

Наличие большого количества оперативки на сервере - это очень хорошо. Память будет использоваться под page cache, для того, чтобы не читать данные с диска при обработке запросов. Это делается прозрачно операционной системой и не учитывается в выводе Peak memory usage.

Посмотреть, сколько используется под page cache, можно программой free или top.
Посмотреть общий объём данных в таблицах типа MergeTree так:

SELECT formatReadableSize(sum(bytes)) FROM system.parts WHERE active

Распределение объёма данных по столбцам:

SELECT database, table, name, formatReadableSize(sum(bytes) AS s) FROM system.columns GROUP BY database, table, name ORDER BY s DESC

При обработке запроса, оперативка дополнительно используется:
- совсем немного, для буферов ввода-вывода;
- немного - на хранение блоков данных для обработки;
- для хэш-таблиц и состояний агрегатных функций при выполнении GROUP BY, IN, JOIN, DISTINCT;
- для временных блоков при сортировке;
и тому подобного.

Именно это выводится в Peak memory usage.

Я попробовал ряд стандартных настроек вроде увеличения max_block_size или max_threads/max_distributed_

processing_threads, но это особо не сказалось на времени выполнения запроса.

max_block_size увеличивать не имеет смысла, значение по-умолчанию достаточно большое. Иногда имеет смысл уменьшить.
max_threads выбирается автоматически по количеству физических процессорных ядер, что является нормальным вариантом по-умолчанию.

Вообще на таблице в 267 млн. записей (по 89 млн. на каждой из 3-х машин) я получаю время обычного селекта по всем данным (select * from dist_table;) равное ~19-20 сек.

Нормальный ли это результат или при правильной настройке можно получить результат лучше?

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

четверг, 21 июля 2016 г., 17:44:01 UTC+3 пользователь Stepan Semiokhin написал:

[man...@gmail.com]

Это зависит от того, сколько данных передаётся по сети, и как быстро это происходит.
Использование сети можно посмотреть с помощью утилиты dstat.
(sudo apt-get install dstat)

Там будут столбцы recv и send.
Откройте отдельный терминал на сервере-инициаторе, запустите dstat и смотрите на значение в столбце send.

На 1Gbit сети, при использовании всего канала, вы будете видеть значение в районе 120 M.

Если по сети передаётся много данных - может быть, это потому, что всё-таки, GROUP BY делается по длинным строкам, а не по ключам, из которых эти строки получены - в этом случае, запрос можно будет поправить, вынеся это из GROUP BY. То есть, "превращение" ключей агрегации в строки, будет идти уже на сервере-инициаторе запроса, и эти данные не придётся передавать по сети.

четверг, 21 июля 2016 г., 18:52:50 UTC+3 пользователь Stepan Semiokhin написал:

[man...@gmail.com]

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

Если перепишете так, чтобы GROUP BY был в том же месте, где SELECT ... FROM table, то всё будет хорошо :)

четверг, 21 июля 2016 г., 18:52:50 UTC+3 пользователь Stepan Semiokhin написал:

[Stepan Semiokhin]

Спасибо, сейчас попробую переписать запрос, все-таки даже 6 сек на 87 млн многовато.

Посмотрел по памяти, все сходится, да, как раз 870 мб получается...

А вот dstat дал интересный результат: при отсутствии запросов значение send равно 6,5 Мб и при выполнении запроса практически не меняется (до 7-7,5 Мб, никакими 120  Мб на 1 Гбит не пахнет), в отличие от recv, который увеличивается с 3,5 Мб до 70-75 Мб.

Получается особо ничего лишнего не передается по сети?

пятница, 22 июля 2016 г., 3:36:10 UTC+3 пользователь man...@gmail.com написал:

[Stepan Semiokhin]

После перемещения всех GROUP BY в подзапрос получил 6,5-8,5 секунд (в зависимости от наличия/отсутствия limit) и 30-40 млн. строк, что очень неплохо. Я так понимаю, на что-то похожее и стоит рассчитывать, да?

[man...@gmail.com]

А вот dstat дал интересный результат: при отсутствии запросов значение send равно 6,5 Мб и при выполнении запроса практически не меняется (до 7-7,5 Мб, никакими 120  Мб на 1 Гбит не пахнет), в отличие от recv, который увеличивается с 3,5 Мб до 70-75 Мб.

Получается особо ничего лишнего не передается по сети?

В данном запросе, сервер инициатор ничего существенного не передаёт, и не должен - поэтому send примерно нулевой. Зато он принимает промежуточные результаты агрегации, и это отображается в recv.

Почему 70 МБ/сек., а не ~120 МБ/сек?
Возможны такие причины:
- сеть работает медленнее 1 Gbit (по факту, в данном случае, на это может влиять много обстоятельств);
- выполнение запроса в это время упирается в CPU;

Подробнее надо смотреть вживую.

Получается особо ничего лишнего не передается по сети?

Почему же. Может быть, передаётся, и можно ещё уменьшить объём передаваемых данных...
Я не вижу окончательного варианта запроса, и не знаю, какие там данные - поэтому не могу быть уверенным на 100%.

[man...@gmail.com]

Я не могу точно сказать, с какой скоростью должен выполняться именно этот запрос - зависит от данных: структуры таблицы, среднего размера строк, коэффициента сжатия данных, распределения уникальных значений, распределения размеров строк в словарях и т. п.

Может быть, ваш запрос может работать со скоростью не 30 млн. строк в секунду, а 300 млн. строк в секунду :)
Если бы я занимался этой задачей, я бы подробно изучил, что и как происходит на каждом сервере, что является узким местом и почему.

пятница, 22 июля 2016 г., 12:05:33 UTC+3 пользователь Stepan Semiokhin написал:

[Stepan Semiokhin]

Вопрос как раз про то, на что база способная в плане пропускной способности на основании исторических кейсов, не сомневаюсь, что в Метрике используются и более сложные запросы, вот интересно, что "в среднем" можно ожидать на запросе такой сложности)

Спасибо большое, узнал все, что хотел и даже больше. Судя по тому, что локально на одной таблице я получаю 14-15 млн. строк/сек. узкое место стоит искать в структуре таблице, буду ковырять в эту сторону, еще раз спасибо!

пятница, 22 июля 2016 г., 15:49:23 UTC+3 пользователь man...@gmail.com написал: