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

Question

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

Согласно документации:

spark.sql.shuffle.partitions 200 Настраивает количество разделов, используемых при перетасовке данных для объединений или агрегатов.

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

Мне интересно, могу ли я сказать спарку просто настроить количество разделов в зависимости от объема данных?Т.е. установить максимальный размер раздела во время операций соединения?

Дополнительный вопрос - как в Spark узнать, каков общий размер наборов данных, которые нужно обработать, при перераспределении на 200 примерно равных разделов?

Заранее спасибо!

Answer 1

AFAIK, нет такой опции, чтобы нацелить разделы в случайном порядке для определенного размера вывода.Так что эта настройка оставлена ​​для вас ... Это в некоторой степени может быть решено на пути чтения вниз по течению при определенном сценарии.Допустим, вы объединяете данные и записываете вывод в паркет через hdfs.Вы можете перераспределить результаты вашего запроса на 1 (или очень небольшое количество разделов).Рассматривайте это как воронку - объединение с некоторым агрегированием, выполненным с 200 разделами, а затем дальнейшее снижение параллелизма по агрегированным данным (это должно включать относительно небольшой ввод-вывод).Предположим, вы нацелены на размер блока 256 МБ.Опции - это вывод где-то вокруг него, намного ниже или намного выше.В первых двух случаях вы в основном достигли того, к чему стремились, и это позволяет избежать слишком фрагментированных данных (а в случае hdfs - слишком большого количества блоков в namenode).Но если ваши выходные данные превышают размер целевого блока, что, очевидно, повлияет на время выполнения последующих заданий, вы можете использовать spark.sql.files.maxPartitionBytes для управления количеством разделов, в которые эти данные считываются.Поэтому, даже если у вас есть выходной файл объемом 2 ГБ, установка этого параметра на 128 МБ приведет к 16 разделам на пути чтения.

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