title: Apache Spark in Scala Stackoverflow with tail recursion
body: <p>The following error occurs:</p>
  
  <pre><code>org.apache.spark.SparkException: Job aborted due to stage failure: Task serialization failed: java.lang.StackOverflowError
  </code></pre>
  
  <p>When I am performing a tail-recursive function using Scala. I was under the impression that tail recursion in Scala can't overflow and that was one of the strengths of Scala.</p>
  
  <p>Method below:</p>
  
  <pre><code>  def gdAll(inputRDD : RDD[(Int, Vector, Int, Vector, Double)]) : RDD[(Int, Vector, Int, Vector, Double)] = {
  
  val step = 0.0000055
  val h4 = 0.05
  
  val errors = inputRDD.map { case (itemid, itemVector, userid, userVector, rating) =&gt;
    (itemid, itemVector, userid, userVector, rating, ((rating - userVector.dot(itemVector)) * itemVector) - h4 * userVector)
  }.cache
  
  val currentRMSE = sqrt(errors.aggregate(0.0)((accum, rating) =&gt; accum + pow(rating._5 - rating._4.dot(rating._2), 2), _ + _) / errors.count)
  
  val totalUserError = errors.aggregate(Vector(0.0, 0.0))((accum, error) =&gt; accum + error._6, _+_)
  
  val usersByKey = errors.map { case (itemid, itemVector, userid, userVector, rating, error) =&gt;
    (userid, (userVector, itemid, itemVector, rating, error))
  }
  
  val updatedUserFactors = usersByKey.map { case ((userid, (userVector, itemid, itemVector, rating, error))) =&gt;
  
    (itemid, itemVector, userid, userVector + (step * totalUserError), rating)
  }
  
  val fullyUpdatedUserFactors = updatedUserFactors.map{ case ((itemid, itemVector, userid, userVector, rating)) =&gt;
    (itemid, itemVector, userid, userVector, rating, ((rating - userVector.dot(itemVector)) * userVector) - h4 * itemVector)}
  
  val itemsByKey = fullyUpdatedUserFactors.map { case (itemid, itemVector, userid, userVector, rating, error) =&gt;
    (itemid, (itemVector, userid, userVector, rating, error))
  }
  
  val totalItemError = fullyUpdatedUserFactors.aggregate(Vector(0.0, 0.0))((accum, error) =&gt; accum + error._6, _+_)
  
  val updatedItemFactors = itemsByKey.map { case (itemid, (itemVector, userid, userVector, rating, error)) =&gt;
  
    (itemid, itemVector + (step * totalItemError), userid, userVector, rating) // totalItemError = itemError
  }
  
  val newRMSE = sqrt(updatedItemFactors.aggregate(0.0)((accum, rating) =&gt; accum + pow(rating._5 - rating._4.dot(rating._2), 2), _ + _) / errors.count)
  
  println("Original RMSE: " + currentRMSE + " New RMSE: " + newRMSE)
  
  val changeInRMSE = (newRMSE - currentRMSE).abs
  
  if (changeInRMSE &lt; 0.0000005) {
  
    return updatedItemFactors
  }
  
  errors.unpersist()
  
  gdAll(updatedItemFactors) // repeat if change is still large
  
   }
  </code></pre>
  
  <p>Any ideas? Thank you.</p>