Я пытаюсь выполнить простое преобразование общих данных обхода, используя хост Spark на EC2, используя это руководство, мой код выглядит следующим образом:
package ccminer
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
object ccminer {
val english = "english|en|eng"
val spanish = "es|esp|spa|spanish|espanol"
val turkish = "turkish|tr|tur|turc"
val greek = "greek|el|ell"
val italian = "italian|it|ita|italien"
val all = (english :: spanish :: turkish :: greek :: italian :: Nil).mkString("|")
def langIndep(s: String) = s.toLowerCase().replaceAll(all, "*")
def main(args: Array[String]): Unit = {
if (args.length != 3) {
System.err.println("Bad command line")
System.exit(-1)
}
val cluster = "spark://???"
val sc = new SparkContext(cluster, "Common Crawl Miner",
System.getenv("SPARK_HOME"), Seq("/root/spark/ccminer/target/scala-2.10/cc-miner_2.10-1.0.jar"))
sc.sequenceFile[String, String](args(0)).map {
case (k, v) => (langIndep(k), v)
}
.groupByKey(args(2).toInt)
.filter {
case (_, vs) => vs.size > 1
}
.saveAsTextFile(args(1))
}
}
И я запускаю его с помощью команды следующим образом:
sbt/sbt "run-main ccminer.ccminer s3n://aws-publicdatasets/common-crawl/parse-output/segment/1341690165636/textData-* s3n://parallelcorpus/out/ 2000"
Но очень быстро он терпит неудачу с ошибками следующим образом
java.lang.OutOfMemoryError: Java heap space
at com.ning.compress.BufferRecycler.allocEncodingBuffer(BufferRecycler.java:59)
at com.ning.compress.lzf.ChunkEncoder.<init>(ChunkEncoder.java:93)
at com.ning.compress.lzf.impl.UnsafeChunkEncoder.<init>(UnsafeChunkEncoder.java:40)
at com.ning.compress.lzf.impl.UnsafeChunkEncoderLE.<init>(UnsafeChunkEncoderLE.java:13)
at com.ning.compress.lzf.impl.UnsafeChunkEncoders.createEncoder(UnsafeChunkEncoders.java:31)
at com.ning.compress.lzf.util.ChunkEncoderFactory.optimalInstance(ChunkEncoderFactory.java:44)
at com.ning.compress.lzf.LZFOutputStream.<init>(LZFOutputStream.java:61)
at org.apache.spark.io.LZFCompressionCodec.compressedOutputStream(CompressionCodec.scala:60)
at org.apache.spark.storage.BlockManager.wrapForCompression(BlockManager.scala:803)
at org.apache.spark.storage.BlockManager$$anonfun$5.apply(BlockManager.scala:471)
at org.apache.spark.storage.BlockManager$$anonfun$5.apply(BlockManager.scala:471)
at org.apache.spark.storage.DiskBlockObjectWriter.open(BlockObjectWriter.scala:117)
at org.apache.spark.storage.DiskBlockObjectWriter.write(BlockObjectWriter.scala:174)
at org.apache.spark.scheduler.ShuffleMapTask$$anonfun$runTask$1.apply(ShuffleMapTask.scala:164)
at org.apache.spark.scheduler.ShuffleMapTask$$anonfun$runTask$1.apply(ShuffleMapTask.scala:161)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:161)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:102)
at org.apache.spark.scheduler.Task.run(Task.scala:53)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:213)
at org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:49)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:178)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:744)
Итак, мой основной вопрос: что нужно для написания задачи Spark, которая может группироваться по ключу с почти неограниченным объемом ввода без исчерпания памяти?
Наиболее распространенная причина исключений java.lang.OutOfMemoryError в задачах тасования (таких как groupByKey, reduceByKey и т.д.) - это низкий уровень parallelism.
Вы можете увеличить значение по умолчанию, установив свойство spark.default.parallelism в configuration.
Итак, это говорит о том, что у вас закончилось выделенное пустое пространство JVM. Вы можете увеличить размер кучи, но все же это ограничено возможностями системы (не может превышать объем физической памяти).
С другой стороны, как объясняется homutov, это происходит в больших операциях сбора. Например, groupByKey, reduceByKey, cartisien + mapToPair. Эти операции собирают данные RDD в одно место, что приводит к тому, что JVM заканчивается из кучного пространства.
Что вы можете сделать?
По моему опыту, когда у кластера/системы есть ограниченные ресурсы, вы можете использовать Руководство по настройке искры. spark.defualt.parallelism может быть увеличено до тех пор, пока вы не сможете сопровождать задачу в своем кластере/системе. [Однажды я выполнил реализацию KNN для 14000 экземпляров, 1024 набора данных объектов на моей виртуальной машине для ноутбука, настроив parallelism],
Command line flag : --conf spark.default.parallelism=4 ; 4 is the parallelism value
Помните, что вам нужно TUNE использовать эти функции для наиболее эффективного и отказоустойчивого (завершения работы кучи), чтобы получить наилучшие результаты из Spark.
Дополнительно
Не используйте примитивные типы данных вместо оберток. И используйте Массивы вместо коллекций.
ex : List<Integers> vs int[] ; int[] is better than List
В Spark-массивах можно сэкономить много ценных мест и повысить производительность.
Также используйте переменные BroadCast вместо декартового произведения или любую большую комбинационную задачу.