title: "Container killed by YARN for exceeding memory limits. 10.4 GB of 10.4 GB physical memory used" on an EMR cluster with 75GB of memory
body: <p>I'm running a 5 node Spark cluster on AWS EMR each sized m3.xlarge (1 master 4 slaves). I successfully ran through a 146Mb bzip2 compressed CSV file and ended up with a perfectly aggregated result.</p>
  
  <p>Now I'm trying to process a ~5GB bzip2 CSV file on this cluster but I'm receiving this error:</p>
  
  <blockquote>
    <p>16/11/23 17:29:53 WARN TaskSetManager: Lost task 49.2 in stage 6.0 (TID xxx, xxx.xxx.xxx.compute.internal): ExecutorLostFailure (executor 16 exited caused by one of the running tasks) Reason: Container killed by YARN for exceeding memory limits. 10.4 GB of 10.4 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.</p>
  </blockquote>
  
  <p>I'm confused as to why I'm getting a ~10.5GB memory limit on a ~75GB cluster (15GB per 3m.xlarge instance)...</p>
  
  <p>Here is my EMR config:</p>
  
  <pre><code>[
   {
    "classification":"spark-env",
    "properties":{
  
    },
    "configurations":[
       {
          "classification":"export",
          "properties":{
             "PYSPARK_PYTHON":"python34"
          },
          "configurations":[
  
          ]
       }
    ]
  },
  {
    "classification":"spark",
    "properties":{
       "maximizeResourceAllocation":"true"
    },
    "configurations":[
  
    ]
   }
  ]
  </code></pre>
  
  <p>From what I've read, setting the <code>maximizeResourceAllocation</code> property should tell EMR to configure Spark to fully utilize all resources available on the cluster. Ie, I should have ~75GB of memory available... So why am I getting a ~10.5GB memory limit error?
  Here is the code I'm running:</p>
  
  <pre><code>def sessionize(raw_data, timeout):
  # https://www.dataiku.com/learn/guide/code/reshaping_data/sessionization.html
      window = (pyspark.sql.Window.partitionBy("user_id", "site_id")
                .orderBy("timestamp"))
      diff = (pyspark.sql.functions.lag(raw_data.timestamp, 1)
              .over(window))
      time_diff = (raw_data.withColumn("time_diff", raw_data.timestamp - diff)
                   .withColumn("new_session", pyspark.sql.functions.when(pyspark.sql.functions.col("time_diff") &gt;= timeout.seconds, 1).otherwise(0)))
      window = (pyspark.sql.Window.partitionBy("user_id", "site_id")
                .orderBy("timestamp")
                .rowsBetween(-1, 0))
      sessions = (time_diff.withColumn("session_id", pyspark.sql.functions.concat_ws("_", "user_id", "site_id", pyspark.sql.functions.sum("new_session").over(window))))
      return sessions
  def aggregate_sessions(sessions):
      median = pyspark.sql.functions.udf(lambda x: statistics.median(x))
      aggregated = sessions.groupBy(pyspark.sql.functions.col("session_id")).agg(
          pyspark.sql.functions.first("site_id").alias("site_id"),
          pyspark.sql.functions.first("user_id").alias("user_id"),
          pyspark.sql.functions.count("id").alias("hits"),
          pyspark.sql.functions.min("timestamp").alias("start"),
          pyspark.sql.functions.max("timestamp").alias("finish"),
          median(pyspark.sql.functions.collect_list("foo")).alias("foo"),
      )
      return aggregated
   spark_context = pyspark.SparkContext(appName="process-raw-data")
  spark_session = pyspark.sql.SparkSession(spark_context)
  raw_data = spark_session.read.csv(sys.argv[1],
                                    header=True,
                                    inferSchema=True)
  # Windowing doesn't seem to play nicely with TimestampTypes.
  #
  # Should be able to do this within the ``spark.read.csv`` call, I'd
  # think. Need to look into it.
  convert_to_unix = pyspark.sql.functions.udf(lambda s: arrow.get(s).timestamp)
  raw_data = raw_data.withColumn("timestamp",
                                 convert_to_unix(pyspark.sql.functions.col("timestamp")))
  sessions = sessionize(raw_data, SESSION_TIMEOUT)
  aggregated = aggregate_sessions(sessions)
  aggregated.foreach(save_session)
  </code></pre>
  
  <p>Basically, nothing more than windowing and a groupBy to aggregate the data.</p>
  
  <p>It starts with a few of those errors, and towards halting increases in the amount of the same error.</p>
  
  <p>I've tried running spark-submit with <strong>--conf spark.yarn.executor.memoryOverhead</strong> but that doesn't seem to solve the problem either.</p>