按键分组时Spark会耗尽内存

scala amazon-ec2 apache-spark

12275 观看

2回复

302 作者的声誉

我试图使用本指南在EC2上使用Spark主机对常见爬网数据进行简单转换,我的代码如下所示:

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任务,可以按键分组几乎无限量的输入,而不会耗尽内存?

作者: John McCrae 的来源 发布者: 2014 年 3 月 25 日

回应 2


14

631 作者的声誉

在shuffle任务中的java.lang.OutOfMemoryError异常(例如groupByKey,reduceByKey等)的最常见原因是低并行度

您可以通过在配置中设置spark.default.parallelism属性来增加默认值。

作者: homutov 发布者: 2014 年 4 月 18 日

4

5951 作者的声誉

所以这说明你已经耗尽了JVM 的已分配堆空间。您可能会增加堆大小,但这仍然受系统功能的限制(不能超过物理RAM的数量)。

另一方面,正如homutov所解释的那样,这发生在大型收集操作中。例如groupByKey,reduceByKey,cartisien + mapToPair。这些操作将RDD数据收集到一个地方,使JVM耗尽堆空间

你能做什么?

根据我的经验,当集群/系统资源有限时,您可以使用Spark调优指南spark.default.parallelism可以增加,直到你可以将任务带入你的集群/系统[我曾经通过调整并行性在我的笔记本电脑的虚拟机上运行了14000个实例,1024个特征数据集的KNN实现]。

Command line flag :   --conf spark.default.parallelism=4   ; 4 is the parallelism value

请记住,你需要TUNE这些功能来最有效和失效避免(耗尽堆)设置以获得最好的结果了星火。

另外

请记住使用use primitive数据类型而不是包装器。并使用数组而不是集合

 ex :  List<Integers> vs int[] ; int[] is better than List 

在Spark数组中可以节省许多宝贵的空间并提高性能。

也可以使用BroadCast变量代替笛卡尔积或任何大型组合任务。

作者: Kavindu Dodanduwa 发布者: 2015 年 7 月 14 日
32x32