Увеличьте доступную память PySpark во время выполнения

Я пытаюсь создать рекомендацию, используя Spark, и просто закончилась нехватка памяти:

Exception in thread "dag-scheduler-event-loop" java.lang.OutOfMemoryError: Java heap space

Я хотел бы увеличить память, доступную Spark, изменив свойство spark.executor.memory в PySpark во время выполнения.

Это возможно? Если да, то как?

Обновление

вдохновленный ссылкой в комментарии @zero323, я попытался удалить и воссоздать контекст в PySpark:

del sc
from pyspark import SparkConf, SparkContext
conf = (SparkConf().setMaster("http://hadoop01.woolford.io:7077").setAppName("recommender").set("spark.executor.memory", "2g"))
sc = SparkContext(conf = conf)

вернулся:

ValueError: Cannot run multiple SparkContexts at once;

Это странно, поскольку:

>>> sc
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
NameError: name 'sc' is not defined

Ответы

Ответ 1

Вы можете установить spark.executor.memory при запуске pyspark-shell

pyspark --num-executors 5 --driver-memory 2g --executor-memory 2g

Ответ 2

Я не уверен, почему вы выбрали ответ выше, когда требуется перезагрузка вашей оболочки и открытие с помощью другой команды! Хотя это работает и полезно, есть встроенное решение, которое действительно запрашивается. Это, по сути, то, что @zero323 упоминается в комментариях выше, но ссылка приводит к сообщению, описывающему реализацию в Scala. Ниже приведена рабочая реализация специально для PySpark.

Примечание. SparkContext, который вы хотите изменить параметры, не должен быть запущен, иначе вам нужно будет закрыть его, изменить настройки и повторно открыть.

from pyspark import SparkContext
SparkContext.setSystemProperty('spark.executor.memory', '2g')
sc = SparkContext("local", "App Name")

Источник: https://spark.apache.org/docs/0.8.1/python-programming-guide.html

p.s. если вам нужно закрыть SparkContext, просто используйте:

SparkContext.stop(sc)

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

sc._conf.getAll()

Ответ 3

Насколько я знаю, во время выполнения невозможно было бы изменить spark.executor.memory. Контейнеры на datanodes будут созданы еще до инициализации искрового контекста.

Ответ 4

Ссылаясь на это, после 2.0.0 вам не нужно использовать SparkContext, но SparkSession с методом conf как SparkSession ниже:

spark.conf.set("spark.executor.memory", "2g")

Ответ 5

Просто используйте опцию config при настройке SparkSession (начиная с 2.4)

MAX_MEMORY = "5g"

spark = SparkSession \
    .builder \
    .appName("Foo") \
    .config("spark.executor.memory", MAX_MEMORY) \
    .config("spark.driver.memory", MAX_MEMORY) \
    .getOrCreate()