Есть ли лучший способ избежать перекоса данных путем объединения в Redshift?

Question

Запрос SQL, как показано ниже, вызывает заполнение одного узла диска кластера Redshift

insert
into
  report.agg_info
  ( pd ,idate ,idate_str ,app_class ,app_superset ,aid ,pf ,is ,camp ,ua_camp_id ,country ,is_predict ,cohort_size ,new_users ,retained ,acc_re ,day_iap_rev ,iap_rev ,day_rev ,rev ) 
select
    p.pd ,
    p.idate ,
    p.idate_str ,
    p.app_class ,
    p.app_superset ,
    p.aid ,
    p.pf ,
    p.is ,
    p.camp ,
    p.ua_camp_id ,
    p.country ,
    1 as is_predict ,
    p.cohort_size ,
    p.new_users ,
    p.retained ,
    ar.acc_re ,
    p.day_iap_rev ,
    ar.iap_rev ,
    p.day_rev ,
    ar.rev
from
    tmp_predict p
join
    tmp_accumulate ar
        on p.pd = ar.pd
        and p.idate = ar.idate
        and p.aid = ar.aid
        and p.pf = ar.pf
        and p.is = ar.is
        and p.camp = ar.camp
        and p.ua_camp_id = ar.ua_camp_id
        and p.country = ar.country

И план запроса:

XN Hash Join DS_DIST_BOTH (cost=11863664.64..218084556052252.12 rows=23020733790769 width=218)
    -> XN Seq Scan on tmp_predict p (cost=0.00..3954554.88 rows=395455488 width=188)
    -> XN Hash (cost=3954554.88..3954554.88 rows=395455488 width=165)
      -> XN Seq Scan on tmp_accumulate ar (cost=0.00..3954554.88 rows=395455488 width=165)

Из приведенного выше изображения мы знаем, что node-39 содержит больше данных, чем другие узлы. Потому что данные искажены на join.

Чтобы решить эту проблему, мы стараемся использовать update вместо join

update
  report.agg_info
set
  acc_re = ar.acc_re,
  iap_rev = ar.iap_rev,
  rev = ar.rev
from
  tmp_accumulate ar
where
  report.agg_info.pd = ar.pd
  and report.agg_info.idate = ar.idate
  and report.agg_info.aid = ar.aid
  and report.agg_info.pf = ar.pf
  and report.agg_info.is = ar.is
  and report.agg_info.camp = ar.camp
  and report.agg_info.ua_camp_id = ar.ua_camp_id
  and report.agg_info.country = ar.country

План запроса

XN Hash Join DS_BCAST_INNER (cost=11863664.64..711819961371132.00 rows=91602 width=254)
    -> XN Seq Scan on agg_info (cost=0.00..2.70 rows=270 width=224)
    -> XN Hash (cost=3954554.88..3954554.88 rows=395455488 width=170)
      -> XN Seq Scan on tmp_accumulate ar (cost=0.00..3954554.88 rows=395455488 width=170)

Данные распределяются равномерно по всем узлам в соответствии с рисунком. Однако в каждом узле больше данных.

Я хочу знать, есть ли лучший метод обработки перекоса данных при объединении в Redshift?

Answer 1

https://docs.aws.amazon.com/redshift/latest/dg/c-analyzing-the-query-plan.html

Найдите следующие операторы широковещательной передачи, где выполняются дорогостоящие операции:
• DS_BCAST_INNER: указывает, что таблица передается на все вычислительные узлы, что хорошо длянебольшая таблица, но не идеальная для таблицы большего размера.
• DS_DIST_ALL_INNER: указывает, что вся рабочая нагрузка находится на одном срезе.
• DS_DIST_BOTH: указывает на интенсивное перераспределение.

DS_DIST_BOTH в вашем первом запросе происходит перераспределение обеих таблиц в определенном столбце. Вы не включили выбранный столбец в свой фрагмент EXPLAIN, но это, вероятно, первый столбец в объединении.

DS_BCAST_INNER передает завершенную копию tmp_accumulate на каждый узел. Обе эти операции довольно дороги и медленны.

Ваше объединение очень широкое, и кажется, что первый столбец довольно искажен. Вы можете попробовать 2 подхода для устранения перекоса и предотвращения широковещания:

1. Измените порядок столбцов, объявленных в объединении, чтобы объявить большинство уникальными (или минимум перекошен) столбец первый. Первый столбец обычно используется в качестве ключа перераспределения. (Примечание: это не будет работать, если в таблицах уже был ключ diststyle.)
2. Рекомендуется . Поскольку объединение является настолько сложным и искаженным, вы можете предварительно вычислить значение хеш-функции для этих столбцов и затем объединить это значение.

--Example of Pre-Calculated Hash
CREATE TEMP TABLE tmp_predict 
    DISTKEY(dist_hash)
AS SELECT FUNC_SHA1(pd||idate::VARCHAR||aid::VARCHAR||pf::VARCHAR
                      ||is::VARCHAR||camp::VARCHAR||ua_camp_id::VARCHAR
                      ||country::VARCHAR) dist_hash
         ,pd ,idate ,aid ,pf ,is ,camp ,ua_camp_id, country
         ,…
     FROM …
;
CREATE TEMP TABLE tmp_accumulate 
    DISTKEY(dist_hash)
AS SELECT FUNC_SHA1(pd||idate::VARCHAR||aid::VARCHAR||pf::VARCHAR
                      ||is::VARCHAR||camp::VARCHAR||ua_camp_id::VARCHAR
                      ||country::VARCHAR) dist_hash
         ,pd ,idate ,aid ,pf ,is ,camp ,ua_camp_id, country
         ,…
     FROM …
;
INSERT INTO report.agg_info
SELECT …
FROM    tmp_predict p
JOIN    tmp_accumulate ar
    ON  p.dist_hash = ar.dist_hash
;

Answer 2

Данные искажены, потому что кажется, что таблица, с которой вы соединяетесь, имеет все свои данные на одном узле. Это может быть либо из-за отсутствия DISTKEY, либо из-за сильного перекоса DISTKEY.

Таким образом, вы можете либо равномерно распределить данные этой таблицы по всем узлам, используя DISTSTYLE EVEN, либо выбрать другой DISTKEY, который имеет большее распределение в данных.