Have you ever seen the following issues in using Spark?

  • Errors building a Spark release
  • Not knowing what Spark parameters are used at run-time
  • Missing external JARs
  • OutOfMemory on driver applications and/or Spark executors
  • Under utilized cluster CPUs
  • Not knowing how to set Spark streaming batch window size
  • Your local OS disk keeps getting heavy IO
  • Not knowing how to collect thread and heap dumps for Spark executors

If you have experienced any of the above, please check out this Spark troubleshooting guide below (applicable to Spark 1.5+). We embedded the SlideShare presentation here and showed two sample tips from the presentation below.

Example slides:

OutOfMemory – GC overhead limit exceeded

15/12/09 19:57:02 WARN scheduler.TaskSetManager: Lost task 175.0 in stage 68.0 (TID 7588, rhel8.cisco.com): java.lang.OutOfMemoryError: GC overhead limit exceeded
        at org.apache.spark.sql.catalyst.expressions.UnsafeRow.copy(UnsafeRow.java:478)
        at org.apache.spark.sql.catalyst.expressions.UnsafeRow.copy(UnsafeRow.java:55)
  • Error indicates too much time is being spent in garbage collection (98% of the total time)
  • Less than 2% of the heap is recovered
  • From ‘top’, often see “1 CPU core fully used at 100%” but no work is done
  • Tuning #1: Increase executor heapsize

      spark-submit … --executor-memory 4096m --num-executors 20 …

  • OR Tuning #2: Increase spark.storage.memoryFraction (fraction of 1)
  • Fraction of Java heap to use for Spark’s memory cache, default 0.6
  • Note: this is deprecated in Spark 1.6.0
  • Example: Running TPCDS query 67 in Spark SQL

      property: spark.storage.memoryFraction=0.6 – causes OOM
      property: spark.storage.memoryFraction=0.8 – completes OK

  • OR Tuning #3: May also try a different GC policy

Ways to give more cores to your Spark job

  • Spark designed for big data
  • More cores and more memory always better (well, until it breaks!)
  • Ways to max out your cluster, example cluster

      40 vCores / node
      128GB memory / node
      5-node

  • Method #1 – Start with evenly divided memory and cores

      --executor-memory 2500m --num-executors 200
      Total # of executors = 200 (default: 1-core each)
      # of executors/node = 40 (fully using all cores)
      Total memory used = 500 GB

  • Method #2 – Heap size non-negotiable

      --executor-memory 6g --num-executors 80
      Total # of executors = 80 (1-core each)
      # of executors/node = 16 (40% CPU utilization)
      Total memory used ~= 500 GB

  • Increase cores per executor as

      --executor-memory 6g --num-executors 80 -–executor-cores 2
      Forcing 80% utilization

  • Compare a sample batch workload

      1-core executors: 611 seconds
      2-core executors: 410 seconds
      33% time reduction!

Join The Discussion

Your email address will not be published. Required fields are marked *