Spark Dataproc StackOverflowError caused by RDD.map()

Viewed 23

I'm trying to run a Spark job on gcloud dataproc. I have been able to run Spark jobs on the cluster before, and the job I'm submitting also starts to execute, but ultimately runs into a StackOverflowError. Here's the job I'm submitting:

gcloud dataproc jobs submit spark --cluster third-cluster --class Main --jars gs://myProject/RoutePlannerCheckpoints.jar --files gs://myProject/timetable.json --region europe-north1 --project myProject -- timetable.json

And here is the code that is throwing the error:

object Main {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().getOrCreate()
    val sc = spark.sparkContext
    sc.setCheckpointDir("hdfs:///myProject/checkpoints/")

    val startTime = System.nanoTime()
    println("Start Time: 0")

    //some other stuff

    passridesGenerator.findPassrides(sc)

    println("Execution time: " + ((System.nanoTime() - startTime) / 1e9d))
  }
}
  def findPassrides(sc: SparkContext): Unit = {
    var timeTableRDD = sc.parallelize(timeTable)

    timeTableRDD = timeTableRDD.sortBy(_.getStartTime)
    println("Timetable: " + timeTableRDD.count())

    var lines = timeTableRDD.groupBy(_.getTrainNumber)
    //lines is now of type RDD[(String, Iterable[CrewRequirement])]
    lines.checkpoint()
    lines.cache()
    println("Lines size: " + lines.count())

    buckets = lines.map(line => {
      var train: Map[String, Bucket] = new HashMap[String, Bucket]
      val crewRequirements = line._2.toIndexedSeq
      for (i <- crewRequirements.indices) {
        println("Entered first loop")
        //some code
        for (j <- i + 1 until crewRequirements.size) {
          //more code
        }
        //even more code
      }
      //returns Map[String, Bucket]
    }).reduce((b1,b2) => {
      //returns Map[String, Bucket]
    })

    //more code follows

And finally, here is the output:

Job [1d58764b592f404dadad343afd822be1] submitted.
Waiting for job output...

...
//lots of logs, no warnings
...

Start Time: 0
22/08/15 09:20:56 WARN org.apache.spark.scheduler.TaskSetManager: Stage 0 contains a task of very large size (9583 KB). The maximum recommended task size is 100 KB.
22/08/15 09:20:59 WARN org.apache.spark.scheduler.TaskSetManager: Stage 1 contains a task of very large size (9583 KB). The maximum recommended task size is 100 KB.
Timetable: 58958
Lines size: 6257
Exception in thread "main" java.lang.StackOverflowError
        at java.lang.reflect.InvocationTargetException.<init>(InvocationTargetException.java:72)
        at sun.reflect.GeneratedMethodAccessor30.invoke(Unknown Source)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1154)
        at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
        at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
        at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
        at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
        at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
...
goes on and on
...

The same program exists as a non-Spark version and works fine on Dataproc (eventually runs out of memory, depending on the provided json's size). The cluster I'm submitting to has a master node n1-standard-8 with 30GB memory and 4 worker nodes n1-standard-4 with 15GB memory. As can be seen in the output, the RDD lines has a bit over 6000 entries, the Iterable inside each entry has on average slightly less than 10 entries. The whole program does indeed need a few gigabytes of memory, however, as mentioned before, the non-Spark version works fine and in the Spark version some of the other input numbers have even been tuned down.

My understanding is that the program fails with the lines.map() function, since this is where it stops debugging (previous print statements are executed). This map function is followed by a reduce() function. The program later continues with additional map and reduce functions (which require more memory than the one that fails). Ultimately, the program should work with significantly more calculations, undoubtedly on a bigger cluster.

I have noticed the warning about the recommended task size being 100KB, but this is unlikely to be achievable without inelegantly dismembering the code. I have tried some things, including checkpoints, even though I have no idea if I used them correctly. My working experience with Spark and Dataproc is near zero. If further information is needed, I will try my best to provide it.

0 Answers
Related