title: Spark runs out of memory when grouping by key
body: <p>I am attempting to perform a simple transformation of common crawl data using Spark host on an EC2 using <a href="http://spark-summit.org/2013/exercises/launching-a-bdas-cluster-on-ec2.html">this guide</a>, my code looks like this:</p>
  
  <pre><code>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) =&gt; (langIndep(k), v)
      }
      .groupByKey(args(2).toInt)
      .filter {
        case (_, vs) =&gt; vs.size &gt; 1
      }
      .saveAsTextFile(args(1))
    }
  }
  </code></pre>
  
  <p>And I am running it with the command as follows:</p>
  
  <pre><code>sbt/sbt "run-main ccminer.ccminer s3n://aws-publicdatasets/common-crawl/parse-output/segment/1341690165636/textData-* s3n://parallelcorpus/out/ 2000"
  </code></pre>
  
  <p>But very quickly it fails with errors as follows </p>
  
  <pre><code>java.lang.OutOfMemoryError: Java heap space
  at com.ning.compress.BufferRecycler.allocEncodingBuffer(BufferRecycler.java:59)
  at com.ning.compress.lzf.ChunkEncoder.&lt;init&gt;(ChunkEncoder.java:93)
  at com.ning.compress.lzf.impl.UnsafeChunkEncoder.&lt;init&gt;(UnsafeChunkEncoder.java:40)
  at com.ning.compress.lzf.impl.UnsafeChunkEncoderLE.&lt;init&gt;(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.&lt;init&gt;(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)
  </code></pre>
  
  <p>So my basic question is, what is necessary to write a Spark task that can group by key with an almost unlimited amount of input without running out of memory?</p>