Pyspark write.csv () закрывается на YARN для превышения ограничений памяти

Question

Предпосылка: я не контролирую свой кластер и работаю над тем, что проблема заключается в моем коде, а не в настройке, используемой моей школой.Может быть, это неправильно, но это предположение лежит в основе этого вопроса.

Почему write.csv () заставляет мое задание pyspark / slurm превышать пределы памяти, когда многие предыдущие операции с большими версиями данных были выполнены успешно, и что я могу с этим сделать?

Ошибка, которую я получаю (много итераций ...):

18/06/02 16:13:41 ERROR YarnScheduler: Lost executor 21 on server.name.edu: Container killed by YARN for exceeding memory limits. 7.0 GB of 7 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.

Я знаю, что могу изменить пределы памяти, но я уже увеличил ее в несколько раз сникаких изменений в результате, и я вполне уверен, что в любом случае я не должен использовать этот объем памяти.Для справки мой призыв к слерму:

spark-submit \
    --master yarn \
    --num-executors 100 \
   --executor-memory 6g \
   3main.py

Так что именно я пытаюсь написать?Ну, я прочитал в 39G .bz2 json, для RDD,

allposts = ss.read.json(filename)

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

abscounts = calculatePosts2(postRDD, sc, spark)
abscounts.printSchema()
print(abscounts.count())

Эти операторы печати работают (вывод ниже).В результате СДР составляет около 60 столбцов на 2000 строк +/-.Эти 60 столбцов включают в себя 1 строку длиной имени subreddit, и 59 удваивается.

root
 |-- subreddit: string (nullable = true)
 |-- count(1): long (nullable = false)
 |-- sum(wordcount): long (nullable = true)
 |-- ingestfreq: double (nullable = true)
 |-- causefreq: double (nullable = true)
 |-- insightfreq: double (nullable = true)
 |-- cogmechfreq: double (nullable = true)
 |-- sadfreq: double (nullable = true)
 |-- inhibfreq: double (nullable = true)
 |-- certainfreq: double (nullable = true)
 |-- tentatfreq: double (nullable = true)
 |-- discrepfreq: double (nullable = true)
 |-- spacefreq: double (nullable = true)
 |-- timefreq: double (nullable = true)
 |-- exclfreq: double (nullable = true)
 |-- inclfreq: double (nullable = true)
 |-- relativfreq: double (nullable = true)
 |-- motionfreq: double (nullable = true)
 |-- quantfreq: double (nullable = true)
 |-- numberfreq: double (nullable = true)
 |-- swearfreq: double (nullable = true)
 |-- functfreq: double (nullable = true)
 |-- absolutistfreq: double (nullable = true)
 |-- ppronfreq: double (nullable = true)
 |-- pronounfreq: double (nullable = true)
 |-- wefreq: double (nullable = true)
 |-- ifreq: double (nullable = true)
 |-- shehefreq: double (nullable = true)
 |-- youfreq: double (nullable = true)
 |-- ipronfreq: double (nullable = true)
 |-- theyfreq: double (nullable = true)
 |-- deathfreq: double (nullable = true)
 |-- biofreq: double (nullable = true)
 |-- bodyfreq: double (nullable = true)
 |-- hearfreq: double (nullable = true)
 |-- feelfreq: double (nullable = true)
 |-- perceptfreq: double (nullable = true)
 |-- seefreq: double (nullable = true)
 |-- fillerfreq: double (nullable = true)
 |-- healthfreq: double (nullable = true)
 |-- sexualfreq: double (nullable = true)
 |-- socialfreq: double (nullable = true)
 |-- familyfreq: double (nullable = true)
 |-- friendfreq: double (nullable = true)
 |-- humansfreq: double (nullable = true)
 |-- affectfreq: double (nullable = true)
 |-- posemofreq: double (nullable = true)
 |-- negemofreq: double (nullable = true)
 |-- anxfreq: double (nullable = true)
 |-- angerfreq: double (nullable = true)
 |-- assentfreq: double (nullable = true)
 |-- nonflfreq: double (nullable = true)
 |-- verbfreq: double (nullable = true)
 |-- articlefreq: double (nullable = true)
 |-- pastfreq: double (nullable = true)
 |-- auxverbfreq: double (nullable = true)
 |-- futurefreq: double (nullable = true)
 |-- presentfreq: double (nullable = true)
 |-- prepsfreq: double (nullable = true)
 |-- adverbfreq: double (nullable = true)
 |-- negatefreq: double (nullable = true)
 |-- conjfreq: double (nullable = true)
 |-- homefreq: double (nullable = true)
 |-- leisurefreq: double (nullable = true)
 |-- achievefreq: double (nullable = true)
 |-- workfreq: double (nullable = true)
 |-- religfreq: double (nullable = true)
 |-- moneyfreq: double (nullable = true)

...

2026

После этого единственная оставшаяся строка в моем коде:

  abscounts.write.csv('bigoutput.csv', header=True)

И это приводит к ошибкам памяти.Это абсолютно не должно занимать места в космосе ... Что я здесь не так делаю?

Спасибо за вашу помощь.

Если вам любопытно / это помогает, весь мой код на github

Answer 1

Во-первых, executor.memoryOverhead отличается от executor-memory.Как вы можете видеть здесь .

С Pyspark, memoryOverhead важен, поскольку он управляет дополнительной памятью, которая может понадобиться python для выполнения некоторых действий (см. здесь )в вашем случае сбор и сохранение файла CSV для каждого раздела.

Чтобы помочь python, вы также можете рассмотреть возможность использования coalesce перед записью.