How to resolve Apache Spark StackOverflowError after multiple unions

Question

I have a Spark Scala program which uses a REST API to get data batch by batch, and once all the data is retrieved I operate on them.

Current Program:

• For each batch, create RDD and merge it with the previous RDD created using the previous API call rdd.union(currentRdd).

• Operate on final RDD

A simple program to reproduce the issue:

    def main(args: Array[String]) = {
     val conf = new SparkConf().setAppName("Union test").setMaster("local[1]")
     val sc = new SparkContext(conf)
     val limit = 1000;
     var rdd = sc.emptyRDD[Int]
     for (x <- 1 to limit) {
       val currentRdd = sc.parallelize(x to x + 3)
       rdd = rdd.union(currentRdd)
     }
     println(rdd.sum())
   }

Problem: - When number of batches are high the program throws a StackOverflowError : Exception in thread "main" java.lang.StackOverflowError at org.apache.spark.rdd.UnionRDD$$anonfun$1.apply

I assume, that when the number of batches increases the RDD dependency graph becomes really complex and throwing the error.

What is the best way to resolve this problem?

Answer 1

已经SparkContext.union知道如何正确计算a union的倍数RDD：

val rdds = List.tabulate(limit + 1)(x => sc.parallelize(x to x + 3))
val rdd = sc.union(rdds)

另外，您可以尝试使用此辅助函数来避免创建unions 的长链：

val rdds = List.tabulate(limit + 1)(x => sc.parallelize(x to x + 3))
val rdd = balancedReduce(rdds)(_ union _)

它应该工作的原因与链接的答案基本相同：s的O(n)链会union破坏堆栈，O(log(n))而union-s的高二叉树则不会。