Кейс оптимизации запросов для Greenplum

Всем привет! Меня зовут Андрей, я работаю дата аналитиком в Data Team продукта Dialog.X5/Insights в X5 Tech. Мы предоставляем аналитику по продажам и покупательскому поведению на данных X5 Group.  Для обработки больших объёмов данных в продукте используется  СУБД (система управления базами данных) Greenplum.

Greenplum – это распределённая СУБД с массово-параллельной архитектурой (Massive Parallel Processing). Она построена на основе PostgreSQL и хорошо подходит для хранения и обработки больших объёмов данных.

В статье рассмотрим ресурсоёмкую операцию для распределённых системCOUNT(DISTINCT)и два способа оптимизации. Для предварительного погружения в планы запросов можно прочитать вот эту хорошую статью.

Список определений, используемых в статье:

Сегмент. Сегменты в Greenplum представляют собой экземпляры PostgreSQL. Каждый сегмент представляет собой независимую базу данных PostgreSQL, где хранится часть данных. Сегмент обрабатывает локальные данные, отдавая результаты мастеру. Мастер, в свою очередь, – это сервер, где развёрнут главный экземпляр PostgreSQL, к которому подключаются клиенты, отправляя SQL-запросы. Сами данные хранятся на серверах-сегментах.

Перераспределение данных. Операция в плане запроса (Redistribute Motion), когда каждый сегмент Greenplum повторно хеширует данные и отправляет строки в разные сегменты в соответствии с хеш-ключом.

Распределение таблицы по полю/списку полей. Хранение таблицы на различных сегментах кластера.Конкретный сегмент для хранения записей выбирается на основе хешей, которые рассчитываются по указанным полям.

Структура данных

Таблица чековых фактов:

fct_receipts (   receipt_id    - идентификатор чека  , receipt_dttm  - дата+время чека , calendar_dk   - числовое представление даты чека например 20240101 , store_id      - идентификатор магазина , plu_id        - идентификатор товара )

Таблица распределена по полю receipt_id достаточно равномерно и партицирована по полю receipt_dttm. Объём данных исчисляется в терабайтах.

Немного о природе данных:

• receipt_dttm для чека уникально;

• receipt_id относится только к одному магазину;

• исходя из вышеперечисленных утверждений, метрика количество чеков аддитивна по времени и по группам магазинов.

Запрос для расчёта количества чеков

Рассмотрим расчёт количества чеков для групп магазинов (это может быть сегментация по регионам, сетям и т. д) и для групп товаров (это может быть сегментация по брендам, производителям и т. д).

На вход запроса подаётся следующий список параметров:

• Период (в статье везде указан август 2023 года)

• Таблица-параметр с группами магазинов selected_stores:

  • store_group_id – группа магазинов

  • store_id – магазин

• Таблица-параметр с группами товаров selected_plu:

  • plu_group_id – группа товаров

  • plu_id – товар

Так как таблицы-параметры небольшого размера относительно таблицы чековых фактов, для таблиц-параметров выбран тип распределенияREPLICATED. Таблицы с распределением REPLICATED дублируются в полном объёме на всех сегментах кластера и при соединении с нимиJOIN происходит локально.

Запрос на расчёт количества чеков в разрезе группы магазинов/товаров выглядит следующим образом:

INSERT INTO receipts_cnt_baskets_draft SELECT   sest.store_group_id , COALESCE(sepl.plu_group_id, 0::INT4) AS plu_group_id , COUNT(DISTINCT fcre.receipt_id)      AS cnt_baskets FROM fct_receipts            AS fcre   INNER JOIN selected_stores AS sest     USING (store_id)   INNER JOIN selected_plu    AS sepl     USING (plu_id) WHERE 1 = 1   AND fcre.receipt_dttm >= '2023-08-01 00:00:00'::TIMESTAMP   AND fcre.receipt_dttm <  '2023-09-01 00:00:00'::TIMESTAMP GROUP BY   GROUPING SETS (     (store_group_id, plu_group_id)   , (store_group_id              )   ) ;

Немного контекста по запросу:

• DISTINCT нужен, так как разные plu_id одной и той же группы товаров могут встречаться в одном чеке.

• GROUPING SETS используется, чтобы не делать несколько обращений к чековой таблице фактов для расчёта разных группировок.

Анализ запроса

На вход запроса подадим параметры:

5 групп магазинов:

35 групп товаров (для сокращения приведён ТОП-5)

Давайте посмотрим на план запроса, построенный оптимизатором GPORCA:

EXPLAIN ANALYZE INSERT INTO receipts_cnt_baskets SELECT   sest.store_group_id , COALESCE(sepl.plu_group_id, 0::INT4) AS plu_group_id , COUNT(DISTINCT fcre.receipt_id)      AS cnt_baskets -- 1 Часть запроса FROM fct_receipts            AS fcre   INNER JOIN selected_stores AS sest     USING (store_id)   INNER JOIN selected_plu    AS sepl     USING (plu_id) WHERE 1 = 1   AND fcre.receipt_dttm >= '2023-08-01 00:00:00'::TIMESTAMP   AND fcre.receipt_dttm <  '2023-09-01 00:00:00'::TIMESTAMP -- 2 часть запроса GROUP BY   GROUPING SETS (     (store_group_id, plu_group_id)   , (store_group_id              )   ) ;

Упрощённый план запроса.  Комментарии к узлам плана пронумерованы.
Читаем снизу вверх:

1 Часть плана (Получение данных)  Итого: Данные подготовлены и лежат на каждом сегменте  по ключу распределения fct_receipts  Shared Scan (share slice:id 4:0)  3) Соединения с таблицами-параметрами (JOIN локальный) ->  Hash Join     Hash Cond: (fct_receipts.plu_id = selected_plu.plu_id)  ->  Hash Join     Hash Cond: (fct_receipts.store_id = selected_stores.store_id)      2) Выборка 1 партиции согласно условию по датам ->  Partition Selector for fct_receipts        Partitions selected: 1      1) Хэширование таблиц параметров ->  Hash     ->  Seq Scan on selected_stores  ->  Hash      ->  Seq Scan on selected_plu

2 Часть плана - расчет COUNT(DISTINCT receipt_id)      Объединение результатов ->  Append  Ключ группировки (store_group_id)     3) COUNT(receipt_id)     ->  HashAggregate       Group Key: share0_ref2.store_group_id            2)  DISTINCT ключ группировки + receipt_id     ->  HashAggregate         Group Key: share0_ref2.store_group_id, share0_ref2.receipt_id                  1) Перераспределение данных по ключу группировки       ->  Redistribute Motion           Hash Key: share0_ref2.store_group_id   Считывание данных из 1 части плана             ->  Shared Scan (share slice:id 1:0)     Ключ группировки (store_group_id, plu_group_id)     3) COUNT(receipt_id)   ->  HashAggregate       Group Key: share0_ref3.store_group_id, share0_ref3.plu_group_id             2) DISTINCT ключ группировки + receipt_id      ->  HashAggregate           Group Key: share0_ref3.store_group_id, share0_ref3.plu_group_id, share0_ref3.receipt_id                  1) Перераспределение данных по ключу группировки         ->  Redistribute Motion             Hash Key: share0_ref3.store_group_id, share0_ref3.plu_group_id             Считывание данных из 1 части плана             ->  Shared Scan (share slice:id 2:0) 

Итого, судя по плану запроса, расчёт количества чеков выполняется в три шага:

1. Перераспределение данных по ключу группировки.

2. DISTINCT ключ группировки + receipt_id.

3. COUNT(receipt_id).

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

Чтобы посмотреть, сколько строк пришло на сегмент или сколько строк обработал сегмент, можно включить параметрSET gp_enable_explain_allstat = ON; передEXPLAIN ANALYZE.

Тогда в плане появится дополнительная информация под каждым узлом:

Путём нехитрого парсинга можно получить список сегментов. В элементах списка последнее значение будет равным количеству строк, обработанных сегментом.

Приведена часть списка сегментов:

Ключ группировки распределился по 58 сегментам, виден явный перекос на одном из сегментов. На 179 сегмент поступило около 269 млн строк, а на 129 сегмент поступило около 31 млн. На 179 сегмент поступило в 9 раз больше строк, а если сравнивать с другими сегментами, то разница будет ещё более ощутимой.

Вышеуказанный запрос выполняется около одной минуты на периоде 1 месяц, в зависимости от нагрузки на кластере.

Оптимизация запроса

Рассмотрим пару вариантов оптимизации такого запроса.

1 вариант. Использование параметра.

Для текущей версии нашего кластера параметр optimizer_force_multistage_agg установлен в off. Значение от версии к версии может изменяться. Для просмотра значения параметра можно воспользоваться командой:

SHOW optimizer_force_multistage_agg;

В документации написано, что данный параметр указывает оптимизатору GPORCA на выбор многоступенчатого агрегатного плана для операций типа COUNT(DISTINCT).
Когда это значение выключено (по умолчанию), оптимизатор GPORCA выбирает между одноэтапным и двухэтапным агрегированным планом в зависимости от стоимости SQL-запроса.

Включаем параметр SET optimizer_force_multistage_agg = on;
Приказываем оптимизатору выбирать двухэтапный агрегированный план.

План на примере ключа группировки (store_group_id, plu_group_id):

Ключ группировки (year_granularity, store_group_id, plu_group_id)  4) COUNT(receipt_id) ->  HashAggregate     Group Key: share0_ref3.store_group_id, share0_ref3.plu_group_id      3) Перераспределение данных по ключу группировки     ->  Redistribute Motion          Hash Key: share0_ref3.store_group_id, share0_ref3.plu_group_id          2) DISTINCT ключ группировки + receipt_id         ->  HashAggregate             Group Key: share0_ref3.store_group_id, share0_ref3.plu_group_id, share0_ref3.receipt_id        1) Перераспределение данных по ключу группировки + receipt_id, receipt_id             ->  Redistribute Motion         Hash Key: share0_ref3.store_group_id, share0_ref3.plu_group_id, share0_ref3.receipt_id, share0_ref3.receipt_id                 ->  Shared Scan (share slice:id 3:0)

В данном плане расчёт количества чеков выполняется в четыре шага:

1. Перераспределение по ключу группировки + receipt_id.
   Это уменьшает перекос, так как количество уникальных значений receipt_id велико.

2. DISTINCT по ключу группировки + receipt_id.
   Это уменьшает количество данных для следующего оператора перераспределения.

3. Перераспределение по ключу группировки.

4. COUNT(receipt_id).

В этом плане двухэтапность выражается в шагах 1, 2. Происходит дополнительное перераспределение иDISTINCTпо ключу группировки + receipt_id.

Вышеуказанный запрос отрабатывает от 3,5 до 4,5 раз быстрее, в зависимости от нагрузки на кластере.

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

Вывод

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

2 вариант. Алгоритмический, расширение ключа группировки.

Метрика количество чеков аддитивна по времени. Можно посчитать количество чеков по дням и потом сделать доагрегацию. Добавление в ключ группировки дня позволит увеличить количество ключей группировки до 30 раз. Если равномерности данных в ключе всё ещё не будет хватать, можно задействовать другие поля таблицы, отвечающие аддитивности метрики количества чеков.

Переписанный запрос:

INSERT INTO receipts_cnt_baskets   WITH draft AS (     SELECT       sest.store_group_id     , fcre.calendar_dk     , COALESCE(sepl.plu_group_id, 0::INT4) AS plu_group_id     , COUNT(DISTINCT fcre.receipt_id)      AS cnt_baskets     FROM fct_receipts            AS fcre       INNER JOIN selected_stores AS sest         USING (store_id)       INNER JOIN selected_plu    AS sepl         USING (plu_id)     WHERE 1 = 1       AND fcre.receipt_dttm >= '2023-08-01 00:00:00'::TIMESTAMP       AND fcre.receipt_dttm <= '2023-09-01 00:00:00'::TIMESTAMP    GROUP BY      GROUPING SETS (        (store_group_id, calendar_dk, plu_group_id)      , (store_group_id, calendar_dk              )   ) ) SELECT   store_group_id , plu_group_id , SUM(cnt_baskets) FROM draft GROUP BY   store_group_id , plu_group_id ;

Для данного запроса оптимизатор выбрал план, как и в начале статьи (на примере ключа группировки (store_group_id, calendar_dk, plu_group_id)):

3) COUNT(receipt_id) ->  HashAggregate     Group Key: share1_ref3.store_group_id, share1_ref3.calendar_dk, share1_ref3.plu_group_id      2) DISTINCT ключ группировки + receipt_id     ->  HashAggregate         Group Key: share1_ref3.store_group_id, share1_ref3.calendar_dk, share1_ref3.plu_group_id, share1_ref3.receipt_id          1) Перераспределение данных по ключу группировки         ->  Redistribute Motion             Hash Key: share1_ref3.store_group_id, share1_ref3.calendar_dk, share1_ref3.plu_group_id             ->  Shared Scan (share slice:id 2:1)

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

Данный запрос выполняется от 7 до 9 раз быстрее по сравнению с изначальным запросом без хинта, в зависимости от нагрузки на кластере. 

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

Вывод

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

Итоги

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

• Если ключ группировки отличается от ключа распределения таблиц, то операция группировки ведёт к перераспределению данных на кластере. В данных ключа группировки может быть перекос, и важно уметь диагностировать такие случаи.

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

Мы погрузились немного в мир Greenplum, рассмотрели, как СУБД выполняет запросы. Узнали о перекосе и методах борьбы с данным явлением. Надеюсь, было полезно и интересно.

Хочу поблагодарить за консультации Даниила Недумова и за помощь в подготовке статьи – Антона Денисова.