Как разрешить накладные расходы G C в pyspark Databricks

Question

Я пытаюсь загрузить файл размером 100 ГБ json в фрейм данных spark и создаю временный вид на него. Затем я запрашиваю данные в этом представлении с помощью запроса

select * from <table_view> limit 1;

Но запрос не завершается, а выдает ошибку Caused by : java.lang.OutOfMemoryError: GC Overhead limit exceeded. Я выполняю свой код в кластере блоков данных. Вот мои данные кластера:

Cluster Mode : standard
Databricks runtime Version : 6.1(Apache Spark 2.4.4)
Worker Type : 56 GB Memory, 16 cores 3 DBU (min worker= 2, max worker = 8) 
Driver Type : 56 GB Memory, 16 cores 3 DBU

Я попытался установить следующие параметры конфигурации, но не добился успеха.

spark.conf.set("spark.executor.memory", '50g')
spark.conf.set('spark.executor.cores', '5')
spark.conf.set('spark.cores.max', '16')
spark.conf.set("spark.driver.memory",'30g')
spark.conf.set("spark.yarn.executor.memoryOverhead",4096)

Я очень новичок в apache spark. Пожалуйста, дайте мне знать, если требуются другие детали.

Answer 1

Скорее всего, у вас недостаточно ресурсов для загрузки всего файла в виде кадра данных. Я вижу два варианта.

1. Создайте временный фрейм данных, ограничив количество строк после прочтения json и создания табличного представления на этом меньшем фрейме данных. Например, если вы хотите прочитать только 1000 строк, сделайте что-то вроде этого: small_df = entire_df.limit(1000), а затем создайте представление поверх small_df.
2. . Вы можете увеличить ресурсы кластера. Я никогда не использовал среду исполнения Databricks, но вам, скорее всего, нужно увеличить количество рабочих узлов. Я бы предложил пойти с большим количеством исполнителей с меньшей памятью, а не с несколькими громоздкими исполнителями. Это даст вам больше параллелизма.

Еще одна вещь. По умолчанию спарк использует 200 случайных разделов. Итак, в вашем случае каждый раздел будет содержать примерно 0,5 ГБ данных. Вероятно, лучше уменьшить эту сторону, увеличив количество разделов (например, 1000). Это можно сделать, установив spark.conf.set("spark.sql.shuffle.partitions", 1000).