OutOfMemoryError в Scala и Спарк

Question

Я использовал этот код для извлечения биграмм из текстового файла:

import org.apache.spark.{SparkContext, SparkConf} 
object DS_E6 { 

    def main(args: Array[String]): Unit = { 
    case class Bigram(first: String, second: String) { 
     def mkReplacement(s: String) = s.replaceAll(first + " " + second, first + "-" + second) 
    } 

    def stringToBigrams(s: String) = { 
     val words = s.split(" ") 
     if (words.size >= 2) { 
     words.sliding(2).map(a => Bigram(a(0), a(1))) 
     } else 
     Iterator[Bigram]() 
    } 

    val conf = new SparkConf() 
     .setMaster("local") 
     .setAppName("bigram") 
     .set("spark.executor.memory", "1g") 

    val sc = new SparkContext(conf) 
     val data = sc.textFile("data/file.txt") 
     val bigrams = data.flatMap { 
     stringToBigrams 
     }.collect() 

     val bigramCounts = bigrams.groupBy(identity).mapValues(_.size) 
     val threshold = 100 
     val topBigrams = bigramCounts.filter(_._2 >= threshold).map(_._1) 
     topBigrams.foreach(println) 
     val replaced = data.map(r => topBigrams.foldLeft(r)((r, b) => b.mkReplacement(r))) 
     val replaced1 = replaced.zipWithIndex() 
     .map { case (line, i) => i.toString + "," + line} 

     replaced1.coalesce(1).saveAsTextFile("data/output.txt") 
    } 
    } 
} 

мой входной файл 45 МБ, когда я запускаю этот код он показывает мне ниже ошибок: (я думаю это связано с Collect())

java.lang.OutOfMemoryError: GC overhead limit exceeded 

at org.spark-project.jetty.util.BlockingArrayQueue.poll(BlockingArrayQueue.java:342) 
at org.spark-project.jetty.util.thread.QueuedThreadPool.idleJobPoll(QueuedThreadPool.java:526) 
at org.spark-project.jetty.util.thread.QueuedThreadPool.access$600(QueuedThreadPool.java:44) 
at org.spark-project.jetty.util.thread.QueuedThreadPool$3.run(QueuedThreadPool.java:572) 

как я могу решить эту проблему?

Answer 1

Возможно, вы действительно не получаете 1 г памяти, которую вы запрашиваете через SparkConf. Причина в том, что когда master = local, Spark-драйвер и исполнитель будут полностью запущены внутри JVM, который запускает ваш код, показанный здесь, который создает SparkContext. К тому времени уже слишком поздно получить больше кучи Java, чем было выделено при запуске JVM. Вы должны добавить, что -Xmx1G arg для команды IntelliJ использует для запуска JVM, который запускает ваш код.

Вы не указали точно, как вы используете свой код в IntelliJ. Но вам нужно будет создать или изменить «Запустить конфигурацию».

В пользовательском интерфейсе IntelliJ в меню панели инструментов «Запустить» выберите команду «Редактировать конфигурации ...». Это вызовет окно, которое показано ниже. Это показывает мою Run Configuration для запуска в «Scala Console». В поле «Параметры VM:» вы должны указать аргумент jvm arg -Xmx1G. В моем случае я запускаю 2,5 ГБ памяти.

Возможно, IntelliJ уже создал для вас конфигурацию запуска, когда вы ранее запускали приложение. Если нет, используйте это окно для создания нового подходящего типа, например. «Scala Console».

Чтобы проверить память в запущенном приложении Scala, используйте следующие команды, чтобы проверить фактическую текущую, максимальную и свободную память jvm, чтобы узнать, действительно ли вы получили запрошенную память.

• sys.runtime.totalMemory()
• sys.runtime.maxMemory()
• sys.runtime.freeMemory()

Answer 2

Когда вы отправляете ваше приложение попытаться увеличить spark.driver.memory (это 512mb по умолчанию) путем установки параметра:

--driver-memory 4g 

Проверьте документацию искровой для получения дополнительной информации об этом виде параметров: https://spark.apache.org/docs/latest/configuration.html