Проблема с памятью pyspark: вызвано: java.lang.OutOfMemoryError: пространство кучи Java

Question

Folks,

Я запускаю код pyspark для чтения 500 МБ файла из hdfs и построения пустой таблицы из содержимого файла

Информация о кластере:

9 датододов 128 ГБ памяти /48 vCore CPU /Node

Конфигурация работы

  conf = SparkConf().setAppName('test') \
                          .set('spark.executor.cores', 4) \
                          .set('spark.executor.memory', '72g') \
                          .set('spark.driver.memory', '16g') \
                          .set('spark.yarn.executor.memoryOverhead',4096 ) \
                          .set('spark.dynamicAllocation.enabled', 'true') \
                          .set('spark.shuffle.service.enabled', 'true') \
        .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
                          .set('spark.driver.maxResultSize',10000) \
                          .set('spark.kryoserializer.buffer.max', 2044) 

    fileRDD=sc.textFile("/tmp/test_file.txt")
    fileRDD.cache
    list_of_lines_from_file = fileRDD.map(lambda line: line.split(" ")).collect()

ошибка

Собрать кусок выплевывает из памяти ошибок.

18/05/17 19:03:15 ERROR client.TransportResponseHandler: Still have 1 
requests outstanding when connection fromHost/IP:53023 is closed
18/05/17 19:03:15 ERROR shuffle.OneForOneBlockFetcher: Failed while starting block fetches
java.lang.OutOfMemoryError: Java heap space

Буду признателен за любую оказанную помощь.

Answer 1

Немного предыстории по этому вопросу

У меня возникла эта проблема, когда я запускаю код через Jupyter Notebook, который работает на edgenode кластера hadoop

Нахождение в Юпитере

так как вы можете передавать код только из Jupyter через клиентский режим (эквивалентно запуску spark-shell из edgenode) драйвером spark всегда является edgenode, который уже упакован другими долго выполняющимися процессами демона, где доступная память всегда меньше, чем память, необходимая для fileRDD.collect() в моем файле

Работал нормально в спарк-подчинении

Я поместил содержимое из Jupyer в файл.py и вызвал то же самое через spark-submit с теми же настройками., он работал в течение нескольких секунд, причина в том, что spark-submit оптимизирован для выбора узла драйвера из одного из узлов, которым требуется память, свободная от кластера.

spark-submit --name  "test_app" --master yarn --deploy-mode cluster --conf spark.executor.cores=4 --conf spark.executor.memory=72g --conf spark.driver.memory=72g --conf spark.yarn.executor.memoryOverhead=8192 --conf spark.dynamicAllocation.enabled=true  --conf spark.shuffle.service.enabled=true --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.kryoserializer.buffer.max=2044 --conf spark.driver.maxResultSize=1g --conf spark.driver.extraJavaOptions='-XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:MaxDirectMemorySize=2g' --conf spark.executor.extraJavaOptions='-XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:MaxDirectMemorySize=2g' test.py

Следующий шаг:

Нашим следующим шагом будет посмотреть, сможет ли блокнот Jupyter отправить искровое задание в кластер YARN, используя Livy JobServer или аналогичный подход.