Спарк OutOfMemoryError при приеме большого входного файла

Question

У меня есть искровой приложение, которое считывает файл с 100 миллионов строк (каждая строка кода, такие как US1.234.567B1) и получает некоторые образцы из него следующим образом:

val codes = sc.textFile("/data/codes.txt") 

    def getPattern(code: String) = code.replaceAll("\\d", "d") 

    val patterns: RDD[(String, Int)] = codes 
    .groupBy(getPattern) 
    .mapValues(_.size) 
    .sortBy(- _._2) 

    patterns 
    .map { case (pattern, size) => s"$size\t$pattern" } 
    .saveAsTextFile("/tmp/patterns") 

Я запускаю это на master = local [*], и он не работает с java.lang.OutOfMemoryError: GC overhead limit exceeded.

Почему?

Я думал, что Spark может обрабатывать любые размеры ввода, если на нем достаточно места на жестком диске.

Answer 1

Длинные Короче вы пытаетесь использовать Спарк антишаблоном:

.groupBy(getPattern) 
.mapValues(_.size) 

, которые легко могут быть выражены, например, как:

codes.keyBy(getPattern).mapValues(_ => 1L).reduceByKey(_ + _).sortBy(_._2, false) 

Я думал, что искра может обрабатывать любой размер ввода.

Обычно это может быть уменьшено до тех пор, пока вы не сделаете это невозможным. group/groupByKey на RDD создают локальные коллекции для каждого ключа. У каждого из них есть память одного исполнителя.

Answer 2

Да, искра может обрабатывать очень большие файлы, но блок параллелизма является исполнителем. «Ошибка в памяти» - это потому, что память исполнительного механизма искры или память искрового драйвера недостаточны. Попробуйте увеличить значение spark.executor.memory и spark.driver.memory, а также настройте количество исполнителей перед отправкой задания.

Вы можете установить эти значения в файле свойств или в SparkConf или непосредственно в командной строке во время искра-submit. Ссылка http://spark.apache.org/docs/latest/configuration.html