Greenplum vs Citus. Часть 2

Архитектура Citus позволяет использовать различные подходы с точки зрения реализации механизма распределения данных и балансировки нагрузки при доступе к этим данным.

Сценарии использования Citus таковы, что наличие возможного перекоса по данным (data skew) не является чем-то необычным. Например, в случае тривиального примера, когда шардам соответствуют данные определенных пользователей (мы упоминали подход с multi-tenant архитектурой в первой части), добиться равномерности распределения в том виде, как это более привычно для Greenplum, будет проблематично. Какие-то шарды будут априори занимать больше места, так как принимают больший объем данных и, возможно, большую нагрузку. Создателям Citus очевидно пришлось учитывать эти особенности и в подходе к решению задачи балансировки нагрузки при таких исходных условиях.

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

Основные понятия ребаланса — это рассмотренные нами в первой статье шарды (shard), группа шардов (shard group), размещение группы шардов (shard group placement) и группа узлов (node group). Группа шардов содержит множество шардов с одинаковыми диапазонами ключей. Размещение группы шардов — это соответствие группы шардов группе узлов, где группа узлов идентифицируется по некоторому целочисленному идентификатору groupid (идентификатор можно узнать из таблицы pg_dist_node).

Говоря про перенос шардов между узлами, имеется ввиду перемещение всей группы шардов. Такая необходимость становится очевидной, если мы вспомним про свойство co-location шардов, чтобы Citus имел возможность эффективного соединения таблиц.

Прежде чем описать алгоритм ребаланса предлагаю рассмотреть:

Стоимость — это оценочная характеристика группы шардов, позволяющая сравнивать группы шардов между собой для решения задачи балансировки. Например, для стратегии by_disk_size это значение принимается равным занимаемому объему данных на диске, однако пользователь может сопоставить этому значению свою характеристику (реализовав свою пользовательскую функцию и сопоставив её с пользовательской стратегией).

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

Утилизация связывает стоимость и емкость следующим образом:

где  — утилизация i-ого узла,  — стоимость j-ой группы,  — емкость i-ого узла.

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

Первое и главное допущение, что алгоритм не ищет "идеального" решения задачи балансировки, а довольствуется некоторым "локально-оптимальным" положением групп шардов. Основная архитектурная документация даёт ссылку на теоретические основы этой и подобных ей задач, но в данном случае нас интересует практическая сторона. Что означает это "локально-оптимальное" положение? Допустим, в нашем кластере такие группы шардов:

Как видно из таблицы, утилизация нод citus-db1 и citus-db2 отличается на 3 порядка:

Так как стратегия по умолчанию by_disk_size значением стоимости считает размер шарда на диске в байтах, что может быть не очень удобным для восприятия при значимом перекосе, для иллюстрации этого перекоса я использовал самописную функцию norm_node_utilization_for_table для некоторого нормирования значения. Это значение рассчитывается как "во сколько раз объем данных на шардах больше/меньше, чем некоторое идеальное среднее значение, исходя из числа нод и общего объема данных". Также нужно добавить, что стоимость для каждого шарда инкрементируется на значение citus.rebalancer_by_disk_size_base_cost, равное по умолчанию 104857600 байт. Это делается для того, чтобы избежать ситуации неправильной балансировки для шардов с малым количеством данных либо пустых. Для теста и наглядности я выставил это значение в 0.

Таким образом, для моего тестового кластера из 2 нод и 4 шардов общая стоимость (total cost) будет равна 671981568 байт, общая ёмкость (total capacity) будет равна 2.

Важный момент для корректной работы алгоритма ребаланса — это пороговые значения, которые позволяют не делать избыточные перемещения, так как алгоритм не учитывает стоимость самого перемещения. То есть, если для достижения балансировки, которая возникла из-за дисбаланса в 10 ГБ, ему необходимо переместить несколько ТБ данных, то он с радостью запланирует такое перемещение. Можно сказать, что без этих настроек цель сбалансировать кластер любой ценой оправдывает средства. По сути, эти значения представляют собой диапазон утилизации ноды, внутри которого нода считается сбалансированной (threshold). Для данного случая диапазон такой: (369589856, 302391712), что равно плюс-минус 10% от средней утилизации на 2 нодах. Таким образом, если значение попадает в этот интервал, то нода считается достаточно сбалансированной.

Также используется и второе пороговое значение improvement_threshold. Чтобы проиллюстрировать необходимость такого порога, в документации приводится следующий пример: допустим, имеются две ноды с такими стоимостями шардов A=200 GB и B=99 GB. Перемещение 100 GB данных шарда с ноды A на ноду B улучшит их положение с точки зрения баланса (A=100 GB, B=199 GB), но выгода от такого перемещения крайне сомнительная, учитывая требуемые на такое перемещение ресурсы (как мы помним, алгоритм не оценивает стоимость перемещений!). Значение этого порога для стратегии by_disk_size задаётся равным 50%. Смысл таков: при оценке целесообразности перемещений учитывается насколько улучшится утилизация ноды, если такое перемещение выполнить. Если показатель утилизации улучшится более чем на 50%, то перемещение планируется к выполнению.

Citus предоставляет возможность получить план без фактического перераспределения шардов, вызвав функцию get_rebalance_table_shards_plan. Эта функция опирается на реализацию алгоритма ребаланса, что даёт возможность поисследовать его на примере вызова для моего тестового кластера из 2 узлов и таблицы на 4 шарда.

Начальное состояние шардов:

Итак, как выглядит на практике сам алгоритм:

По итогу работы алгоритм в моём случае выбрал следующие шаги для ребаланса dist_table:

Таким образом, после перемещения групп шардов кластер бы сбалансировался следующий образом:

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

Оба варианта опираются на создание новых таблиц перед переносом данных. После успешного завершения перемещения шардов таблицы-источники будут удалены. Если где-то возникнет ошибка, то уже проделанные изменения остаются как есть. У неблокирующего переноса также есть важная деталь, что если у распределенной таблицы отсутствует первичный ключ (или уникальный индекс), то UPDATE- и DELETE-выражения будут завершаться с ошибками. Что в некотором смысле приближает его к блокирующему.

Описание деталей переноса потянет на полную статью, так что я обозначу основные особенности:

Если сравнивать эту часть Citus с Greenplum, то различия принципиальные — для Greenplum сценарии использования с перекосом данных между сегментами возможны (препятствовать он этому не может), но являются скорее исключением и показателем "плохого" распределения данных. Для равномерного распределения данных рекомендуется использовать определенные практики. Выражение "скорость каравана равна скорости самого медленного верблюда" хорошо отражает возможные проблемы, которые возникают в случае перекоса данных. Для контроля за этими параметрами системы есть специальные представления.

Число сегментов в общем случае на всех узлах одинаково. В случае использования зеркалирования (а для продакшен это является обязательным) в Greenplum реализованы 2 стратегии размещения зеркал: group- и spread-зеркалирования.

Таким образом, как таковой задачи ребаланса данных таблиц в понимании Citus в Greenplum нет, так как единицей масштабирования выступает отдельный экземпляр PostgreSQL (то есть сегмент), который содержит порции данных всех таблиц. Однако кластер Greenplum в силу объективных причин (например, ручной перенос зеркал, переключение на зеркала, вывод или ввод в строй новых узлов) может прийти в несбалансированное состояние именно по числу сегментов на узлах (как primary-, так и mirror-сегментам). И тогда кластер нужно приводить в сбалансированное состояние.

И тут время для небольшого спойлера: мы планируем решать эту задачу в рамках Greengage и, конкретнее, в задаче ggshrink кластера (вернее, как части ggshrink).

Как известно, в Greenplum контроль за потреблением ресурсов реализован на базе ресурсных групп (и в более ранних версиях — слотов). Без этого механизма задача распределения ресурсов между множеством потребителей с разными уровнями критичности выполнения задач представляется тpyднopeшaeмой. Очевидно, что эта функциональность must-have для корпоративных пользователей.

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

Формально, для Citus готовых реализаций тестов TPC-DS нет (по крайней мере, мне их найти не удалось). Разработчики рекомендуют к использованию HammerDB. Однако для OLAP это реализация TPC-H спецификации, кроме того, в HammerDB нет реализации для Greenplum.

В рамках данного исследования было интересно сравнить производительность с Greenplum по TPC-DS тестам. Пришлось частично адаптировать тест, который подходит и для Postgres. Небольшими доработками удалось адаптировать его для Citus. В частности, под Citus нужно создать distributed- и reference-таблицы. Создания индексов, которые в этой реализации под PostgreSQL объявляеются для колонок, я оставил как есть, хотя это существенно замедлило загрузку данных, по сравнению с Greenplum, где схема задаёт только ключи распределения и индексы отсутствуют. Я исходил из следующей идеи — раз Citus заявляется как способ масштабирования PostgreSQL, то и подход с поднятием нескольких экземпляров и организация кластера Citus должна соотвествовать этому видению. Наличие существующих индексов также соответствует этой идее. Осталось только разобраться с разделением на distributed- и reference-таблицы и решить, какие ключи использовать для распределения данных по шардам. Я взял разбиение на эти типы таблиц и колонки распределения из предлагаемого разработчиками самого Citus варианта под TPC-DS (это не готовая реализация, а скорее набор скриптов).

Однако при запуске тестов приключилась занятная история. Пристрелочный тест с набором данных около 10 ГБ (в терминах используемой реализации — scaling factor равен 10) для Citus в ряде тестов либо не завершился в разумный интервал времени, либо показал настолько медленное время, что прогон на каком-то весомом объеме данных не имел никакого смысла. Ради интереса я запустил эти тесты отдельно, но отменил их примерно через 3 часа ожидания.

Создателями реализации теста рекомендуется использовать scaling factor равным 3000 (т.е. около 3 ТБ), но с учетом времени выполнения даже сравнительно быстрых тестов я такой набор решил не использовать даже для относительно быстрых тестов. Для анализа показателей времени с точки зрения разницы и для разбора причин этой разницы во времени достаточно и этого набора. Также пришлось исключить 18 запросов из 99, которые с таким вариантом колонок распределения не выполнились из-за ошибки. Подбирать другие колонки и затачивать запросы под конкретные тесты, скорее всего, могло бы нарушить работоспособность других тестов, а исправлять по кругу другие тесты у меня не было никакого желания.

Citus

Greenplum

Со схемой данных и, в целом, со спецификацией TPC-DS тестов можно ознакомиться тут.

С содержанием каждого запроса можно ознакомиться тут. Постфикс в имени теста соотвествует постфиксу имени шаблона. Например, tpcds.01 соответствует шаблон query1.tpl.

Предлагаю детальнее рассмотреть один из тестов, в которых разница производительности была наибольшей и выполнение которых я остановил по тайм-ауту:

Сам запрос:

Особенность данного запроса во вложенном связанном подзапросе. Для каждой строки из табличного выражения customer_total_return нужно определить, является ли значение ctr_total_return для пары покупатель-магазин (sr_customer_sk и sr_store_sk) больше, чем среднее значение для магазина, плюс 20% к этому среднему. Таким образом, в лоб эта задача решается перебором всего множества вычисленного табличного выражения ctr_total_return и сравнением значение атрибута ctr_total_return с результатом агрегирующей функции SUM для данного магазина.

Судя по плану, который выдаёт Citus, он именно так и делает.

План EXPLAIN для Citus:

Как видно из плана, Citus принял решение оформить табличное выражение customer_total_return как распределенный подплан (маркер 1). Внутри план был распределён по 32 таскам исходя из числа шардов таблицы store_returns по умолчанию равным 32 (маркер 2). Выбор соединения Nested Loop (маркер 3), видимо, справедлив для данного случая, так как текущая TPC-DS реализация строит индексы по некоторым полям и такой индекс и был выбран для соединения с таблицей date_dim (маркер 4).

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

Этой части соответствует такой EXPLAIN-план выполнения:

и фактический EXPLAIN ANALYZE:

Как видно, планировщику пришлось принимать решение на некорректных входных данных. В отсутствии других вариантов (так как используется оценка для функции read_intermediate_result) данный узел Function Scan on read_intermediate_result intermediate_result оценил, что вернёт rows=333 строк (маркер 2 в выводе EXPLAIN). На самом деле, судя по EXPLAIN ANALYZE, строк на один порядок больше (маркер 3 в выводе EXPLAIN ANALYZE). Для другого scale factor ошибка будет больше.

Самая медленная часть плана — это как раз Nested Loop (маркер 1) по результатам выполнения подплана. Для каждой строки из промежуточного результата вызывается подсчёт агрегирующей функции (маркер 2), чтобы рассчитать значение SUM для данного ctr_store_sk (маркер 4). Выглядит не слишком оптимальным.

Greenplum и GPORCA, как мне представляется, строят более продвинутый план, который и приводит к такой существенной разнице во времени:

С точки зрения близких частей плана, можно отметить, что распределенной, параллельной части плана Citus можно поставить соответствие Shared Scan (маркер 1). По факту, это то же самое действие и результат с точки зрения данных. У Greenplum в данной реализации TPC-DS в распоряжении какие-либо индексы отсутствуют, все таблицы DISTRIBUTED BY, поэтому таблицу date_dim приходится перераспределять через Broadcast Motion на все сегменты (маркер 2). Затем таблица store_returns соединяется с date_dim, агрегируется по столбцам sr_customer_sk и sr_store_sk и перераспределяется по сегментам по этим двум столбцам (маркер 3).

Ключевое отличие заключается в том, как Greenplum обрабатывает вложенный коррелированный подзапрос.

Во-первых, осуществляется группировка по полю sr_store_sk и предрасчет фукнции SUM (маркер 6).

Во-вторых, это множество строк перераспределяется Broadcast Motion опять-таки по атрибуту sr_store_sk (маркер 5). Это позволит сделать локальный Hash Join исходного множества из Shared Scan с уже предрассчитанным множеством по колонке sr_store_sk и таким образом просто применить условие Join Filter по ctr_total_return с учетом выражения avg(ctr_total_return) * 1.2 (маркер 4).

Этот шаг позволяет выполнить запрос на несколько порядков быстрее. Citus вычисляет агрегат в рамках вложенного корреллированного запроса для каждой строки из inner-множества для соединения Nested Loop. Greenplum все рассчитал и просто осуществляет Hash Join с тем же условием.

Ради интереса я сделал детализацию первого запроса. Тут четко видно, где Citus проводит больше всего времени.

Для себя я сделал следующие выводы: