Почему Spark записывает данные в разные размеры и разное время?

Question

Рассмотрим общий алгоритм:

val first : DataFrame = ...            // about 100 Mb
val second : DataFrame = ...           // about 5 GMb 
val third : DataFrame = ...            // about 7 GMb
val fourth : DataFrame = ...           // about 13 GMb

//all dataframe are filtered, renamed all columns. A new colun is added into `third` and `fourth`  

val = firstAndSecond = first.join(first("first_id") === second("second_id"))
val = thirdsAndAll = firstAndSecond.join(firstAndSecond("some_id") === third("third_id"))
val = fourthAndAll = thirdsAndAll.join(thirdsAndAll("other_id") === fourth("fourth_id"))

fourthAndAll.write.mode(saveMode = SaveMode.Overwrite).parquet("file://C:path")

Примечания *

1. Все кадры данных считываются и записываются с SSD-диска.
2. Операция чтения и записи, выполняемая в / в файлах паркета
3. Программа была запущена на Threadripper с 8 ядрами (16 виртуальных), 80 ГБ ОЗУ (искра потребляла около 25 ГБ), также 99% времени (кроме ситуации, когдапоследний файл записывается) все 16 ядер загружены на 100%.

Проблема

У меня очень разные размеры в выходных файлах паркета от 100 кб до 500 нб,Также большие файлы имеют очень много времени для записи. Например, каждый файл записывается некоторыми потоками, которые записывают файлы размером 500 МБ, 450 МБ (и т. Д.), Выполняющие свою работу слишком долго. (для 500Мб это было около 8 часов)

Есть мысли, как настроить спарк для записи файла паркета с более или менее равным размером загрузки процессора?

Answer 1

(я работаю в основном с RDD s, не имею большого опыта работы с dataframe api, но я думаю, что он работает так же)

Я думаю, .parquet создает файл на раздел rdd. Если размеры файлов резко отличаются, то данные распределяются неравномерно по разделам. Это также может снизить производительность операции join. repartition команда должна помочь.