YARN killing containers for exceeding memory limits
<p>I am running into an issue where YARN is killing my containers for exceeding memory limits:</p>

<pre><code>Container killed by YARN for exceeding memory limits. physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.
</code></pre>

<p>I have 20 nodes that are of m3.2xlarge so they have:</p>

<pre><code>cores: 8
memory: 30
storage: 200 gb ebs
</code></pre>

<p>The gist of my application is that I have a couple 100k assets for which I have historical data generated for each hour of the last year, with a total dataset size of 2TB uncompressed. I need to use this historical data to generate a forecast for each asset. My setup is that I first use s3distcp to move the data stored as indexed lzo files to hdfs. I then pull the data in and pass it to sparkSql to handle the json:</p>

<pre><code> val files = sc.newAPIHadoopFile("hdfs:///local/*",
  classOf[com.hadoop.mapreduce.LzoTextInputFormat],classOf[org.apache.hadoop.io.LongWritable],
  classOf[org.apache.hadoop.io.Text],conf)
val lzoRDD = files.map(_._2.toString)
val data = sqlContext.read.json(lzoRDD)
</code></pre>

<p>I then use a groupBy to group the historical data by asset, creating a tuple of (assetId,timestamp,sparkSqlRow). I figured this data structure would allow for better in memory operations when generating the forecasts per asset.</p>

<pre><code> val p = data.map(asset =&gt;  (asset.getAs[String]("assetId"),asset.getAs[Long]("timestamp"),asset)).groupBy(_._1)
</code></pre>

<p>I then use a foreach to iterate over each row, calculate the forecast, and finally write the forecast back out as a json file to s3.</p>

<pre><code> p.foreach{ asset =&gt;
  (1 to dateTimeRange.toStandardHours.getHours).foreach { hour =&gt;
    // determine the hour from the previous year
    val hourFromPreviousYear = (currentHour + hour.hour) - timeRange
    // convert to seconds
    val timeToCompare = hourFromPreviousYear.getMillis
    val al = asset._2.toList

    println(s"Working on asset ${asset._1} for hour $hour with time-to-compare: $timeToCompare")
    // calculate the year over year average for the asset
    val yoy = calculateYOYforAsset2(al, currentHour, asset._1)
    // get the historical data for the asset from the previous year
    val pa = asset._2.filter(_._2 == timeToCompare)
      .map(row =&gt; calculateForecast(yoy, row._3, asset._1, (currentHour + hour.hour).getMillis))
      .foreach(json =&gt; writeToS3(json, asset._1, (currentHour + hour.hour).getMillis))
  }
}
</code></pre>

<ul>
<li>Is there a better way to accomplish this so that I don't hit the memory issue with YARN? </li>
<li>Is there a way to chunk the assets so that the foreach only operates on about 10k at a time vs all 200k of the assets? </li>
</ul>

<p>Any advice/help appreciated!</p>