OutOfMemoryError: пространство кучи Java в Spark

Question

У меня проблемы с памятью, но я не могу ее решить.Любая помощь высоко ценится.Я новичок в функциональности Spark и pyspark и пытаюсь прочитать большой JSON-файл размером около 5 ГБ и построить rdd, используя

df = spark.read.json("example.json")

Каждый раз, когда я запускаю приведенное выше утверждение, я получаю следующую ошибку:

java.lang.OutOfMemoryError : Java heap space

Мне нужно получить данные JSON в форме RDD, а затем использовать SQL Spark для манипулирования и анализа.Но я получаю ошибку на первом шаге (чтение JSON).Мне известно, что для чтения таких больших файлов необходимы изменения в конфигурации Spark Session.Я следовал за ответами, данными Apache Spark: задание прервано из-за сбоя этапа: «TID x завершился неудачно по неизвестным причинам» и Spark java.lang.OutOfMemoryError: Пространство кучи Java

Я пытался изменить конфигурацию своей SparkSession, но, возможно, неправильно понял некоторые настройки.Ниже приведены мои настройки свечей.

spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.config("spark.memory.fraction", 0.8) \
.config("spark.executor.memory", "14g") \
.config("spark.driver.memory", "12g")\
.config("spark.sql.shuffle.partitions" , "8000") \
.getOrCreate()

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

Answer 1

Попробуйте использовать:

df = spark.read.json("example.json").repartition(100)

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

Мое предложение состоит в том, чтобы уменьшить spark.sql.shuffle.partitions значение до минимума и попробуйте использовать переразметку или параллелизм для увеличения раздела ваших входных / промежуточных кадров данных.

spark = SparkSession \
  .builder \
  .appName("Python Spark SQL basic example") \
  .config("spark.memory.fraction", 0.8) \
  .config("spark.executor.memory", "14g") \
  .config("spark.driver.memory", "12g")\
  .config("spark.sql.shuffle.partitions" , "800") \
  .getOrCreate()