Объединение большого количества фреймов данных

#scala #apache-spark #join #out-of-memory

#scala #apache-spark #Присоединиться #нехватка памяти

Вопрос:

У меня есть массив фреймов данных, к которым я присоединяюсь, используя столбец «id» для создания большего фрейма данных

 val dfArray = Array.fill[DataFrame](n)(spark.emptyDataFrame)
dfArray(0) = df_0
dfArray(1) = df_1
...
dfArray(n) = df_n

val output = dfArray.reduceLeft(_.join(_, Seq("id")))

Если я попытаюсь выполнить какое-либо действие (например: output.show ) на output , это займет действительно много времени. Иногда происходит сбой с ошибками ООМ. Однако, если я записываю output в файл и считываю его обратно в другую переменную, он работает нормально.

У кого-нибудь есть идеи, почему это так и как это решить.

Ответ №1:

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

1. Спасибо за ваш ответ. Я имел в виду действие, show выполняемое над двумя фреймами данных. Если я выполняю show вывод, это занимает несколько минут. Однако, если я запишу вывод в файл (например: output.wirte.csv(path) ) и загружу его обратно в другую переменную (например: val output2 = spark.read.csv(path) ), а затем выполню show над output2 , это займет всего несколько секунд. Я ожидал, что обе операции отображения будут занимать одинаковые промежутки времени, учитывая, что output это кэшируется в памяти. Но это не так.

2. На самом деле, после того, как вы сделали dfArray.reduceLeft(_.join(_, Seq("id"))) , еще ничего не сделано (spark ленив). Даже если вы добавили .cache() . Для show работы spark должен выполнить объединения (по крайней мере, достаточное количество, чтобы иметь возможность печатать первые строки), вот почему это требует времени. Напротив, как только df записан на диск, демонстрация образца не требует усилий.

3. Что касается вашего вопроса о ООМ, который возникает только тогда, когда вы не записываете данные первыми, у меня есть идея, но мне нужно было бы узнать больше о вашем бэкэнде (где находится кластер и какой тип хранилища вы используете).

4. @Oli в настоящее время это не выполняется в кластере. Я запускаю это локально для тестирования.

5. Вы должны предоставить более подробную информацию и весь код, который вы тестируете