Оптимизация в pyspark

Question

Я пишу код в pyspark, где я делаю фильтрацию, объединение, unionall, groupby.

Но я заметил, что перед групповым отключением функция count() занимает 9 минут, а после groupby функция count() занимает 35 минут. Мне нужно несколько советов о том, как оптимизировать предложение groupby, чтобы сократить время обработки.

Параметры конфигурации: исполнитель памяти 12G Num-исполнителей 50 исполнитель сердечники памяти 5 Драйвер 40g размер данных около 1 ТБ

Фрагмент кода, например:

dataframeA = hc.sql("select * from tableA") 



dataframeB = hc.sql("select * from tableB") 

dataframeC = hc.sql("select * from tableC") 



dataframeD = dataframeA.unionAll(dataframeB) 



dataframeE = dataframeD.join(dataframeC, col1 == col2, 'left_outer') 



dataframeF = dataframeE.groupby ([col1, col2] 
        .agg({ 
         max("col3"), 
         sum("col4")/ sum("col5"), 
         ... 
         }) 

предложения?

Answer 1

Вы можете использовать reduceByKey вместо groupByKey.

groupByKey будет перемешать все данные между кластерами и потреблять много ресурсов, но reduceByKey сначала уменьшит данные в каждом кластере, а затем перетасовывает данные.

Answer 2

Сама логика выглядит нормально. Есть несколько вещей, которые вы можете попробовать:

У вас есть соединение и группа, которые обычно предлагают немного перетасовки. Вы можете попытаться уменьшить количество исполнителей, предоставляя больше памяти и ядра.

В вашей группе вы используете клавишу [col1, col2]. Являются ли эти те же столбцы, что и в соединении? Если это так, то они по сути являются одним и тем же столбцом с разницей, когда на левом фрейме нет ключа. Поэтому вы можете использовать только col2, и, надеюсь, оптимизатор улучшит ваш второй случайный выбор (используйте оригинальную перетасовку).