title: Optimization problem on a large volume of data : java.lang.OutOfMemoryError: GC overhead limit exceeded
body: <p>I have a code optimization problem, indeed I work on very large volumetrics. When I run the code on a small volume of data run code without problems but as soon as I go to real size I get the following errors: java.lang.OutOfMemoryError: GC overhead limit exceeded
  and
  java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]</p>
  
  <p>I tried to optimize this already (delete unusefull action ...) , i tried to run my code inside a for loop to run the code sequentially, i tried inside a for each. </p>
  
  <p>I run on "  --num-executors 6 --executor-memory 22G --executor-cores 8 --driver-cores 6 --driver-memory 10G  "</p>
  
  <p>...</p>
  
  <p>This is the main loop of the code </p>
  
  <pre><code>datatest
    .select("yearWeek", "cluster").distinct().collect().toList
    .map(cluster =&gt; (cluster.getString(0), cluster.getString(1)))
    .map(cluster =&gt; {
      TestModel(datatest.filter($"cluster" === cluster._2), cluster._1, cluster._2, ModelCcr_1, ModelErlang_1)
    })
  </code></pre>
  
  <p>...</p>
  
  <p>And inside the function : </p>
  
  <p>... </p>
  
  <pre><code>def TestModel(Test_Filtered: DataFrame, yearWeek: String, Cluster: String, ModelCcr: List[(String, Array[loadModelWithId.DecisionTreeModelElsa])], ModelErlang: List[(String, Array[loadModelWithId.DecisionTreeModelElsa])] ): DataFrame = {
  
  val sparkContext = sparkSession.sparkContext
  val sqlContext = sparkSession.sqlContext
  import sqlContext.implicits._
  
  val initialSetReduce = List(0.0)
  val addToSetReduce = (s: List[Double], v: Double) =&gt; if (s.head==0.0) List(v) else v :: s
  val mergePartitionSetsReduce = (p1: List[Double], p2: List[Double]) =&gt; p1 ++ p2
  
  
  
  val ModelCcr_Off = ModelCcr.filter(x =&gt; x._1 == Cluster)
  val ModelErlang_Off = ModelErlang.filter(x =&gt; x._1 == Cluster)
  
  
  val moyenne_ccr = Test_Filtered.select("ccr").agg(mean("ccr")).map(x =&gt; x.getDouble(0)).head()
  
  val moyenne_erlang = Test_Filtered.select("erlang").agg(mean("erlang")).map(x =&gt; x.getDouble(0)).head()
  
  
  val TrainForDeploy = Test_Filtered
    .select(
    col("erlang")
    , col("resource")
    ,col("resourceIndexCluster")
    , col("dayTime")
    , col("nbConnections").cast("Double")
    , col("ccr")
    , col("resourceType")
    , col("yearWeek")
    ).rdd.map(x =&gt; (
    (x.getDouble(0), Vectors.dense(x.getLong(2),x.getDouble(3), x.getDouble(4), 1D, 1D)),
    (x.getDouble(5), Vectors.dense(x.getLong(2),x.getDouble(3), x.getDouble(4), 1D, 1D))
    , x.getString(6)
    , x.getString(7)
    ,x.getString(1)
  )).persist(StorageLevel.MEMORY_AND_DISK)
  
  
  
  val TrainDeployCCR = TrainForDeploy
    .map(data =&gt;
      ModelCcr_Off.map(x =&gt; x._2
        .map(tree =&gt;
          //Calcul des prédictions pour le ccr
          ((data._2._2, data._2._1, data._3, data._4,data._5), Distrib.noeudPredictPredQuality(tree.decisionTreeModel, data._2._2))
        )))
    .flatMap(x =&gt; x.flatten(x=&gt;x))
  
  
  
  val TrainDeployErlang = TrainForDeploy
    .map(data =&gt;
      ModelErlang_Off.map(x =&gt; x._2
        .map(tree =&gt;
          //Calcul des prédictions pour le ccr
          ((data._1._2, data._1._1, data._3, data._4,data._5), Distrib.noeudPredictPredQuality(tree.decisionTreeModel, data._1._2))
        )))
    .flatMap(x =&gt; x.flatten(x=&gt;x))
  
  
  TrainForDeploy.unpersist()
  
  
  val TrainNodes_erlang = TrainDeployErlang
    .aggregateByKey(initialSetReduce)(addToSetReduce, mergePartitionSetsReduce) 
  
  
  
  val TrainNodes_ccr = TrainDeployCCR
    .aggregateByKey(initialSetReduce)(addToSetReduce, mergePartitionSetsReduce)
  
  
  
  
  val quantile_ccr = TrainNodes_ccr.map(x =&gt; (x._1, x._2)) // enelever calcul des quantile mettre hors de la fonction
    .map(x =&gt; (x._1._5, x._1._1(1), x._1._2, x._1._3, x._1._4, x._2))
    .toDF("resource", "dayTime", "ccr", "resourceType", "yearWeek", "ListPredCCR")
    .withColumn("erreurdelamoyenneCCR", erreurDeLaMoyenne(col("ccr"), col("ListPredCCR"), lit(moyenne_ccr)))
    .withColumn("erreurAuCarreCCR", erreurAuCarre(col("ccr"), col("ListPredCCR")))
  
  
  
  
  val quantile_erlang = TrainNodes_erlang.map(x =&gt; (x._1, x._2)) //a changer le .first
    .map(x =&gt; (x._1._5, x._1._1(1), x._1._2, x._1._3, x._1._4, x._2))
    .toDF("resource", "dayTime", "erlang", "resourceType", "yearWeek", "ListPredErlang")
    .withColumn("erreurdelamoyenneErlang", erreurDeLaMoyenne(col("erlang"), col("ListPredErlang"), lit(moyenne_erlang)))
    .withColumn("erreurAuCarreErlang", erreurAuCarre(col("erlang"), col("ListPredErlang")))
  
  
  
  
  val cluster_Data = quantile_ccr.coalesce(1).join(quantile_erlang.coalesce(1), Seq("resource", "dayTime", "yearWeek", "resourceType"), "inner")
  
  
  
  cluster_Data
  </code></pre>
  
  <p>}</p>
  
  <p>...</p>
  
  <p>Actually the code run 5 weeks easily but for the 6th week the code crash and print GC exception error or TimedOut Error.</p>
  
  <p>I think it's possible that spark's executors have'nt enough memory to run all data ?</p>
  
  <p>I have already try to put some persist or cache function but no success.</p>
  
  <p>Can we free memory after each iteration ? Or if you have another way to optimize the code.</p>
  
  <p>Thanks you for your help.</p>